Adds get events and get hashes

.github/workflows/ci.yml

@@ -21,9 +21,6 @@ jobs:
           - command: check --locked --all
           - command: clippy --locked --all --tests -- -D warnings
           - command: fmt --all -- --check
-          - command: test --locked --all
-          - command: test --no-default-features --locked --all
-          - command: test --all-features --locked --all
           - command: doc --all-features --locked --no-deps
           - command: bench --no-run --locked --all
     steps:
@@ -36,21 +33,25 @@ jobs:
           targets: wasm32-unknown-unknown
       - uses: Swatinem/rust-cache@v2
 
-      # For test commands, we need sigma-node available
-      - name: Build sigma-node
-        if: contains(matrix.command, 'test')
-        run: |
-          # Clone and build sigma-node directly
-          git clone git@github.com:essential-contributions/sigma-node.git
-          cd sigma-node
-          cargo build --release --bin sigma-node
-          # Add to PATH
-          echo "$PWD/target/release" >> $GITHUB_PATH
-          cd ..
-
       - name: Run cargo command
         run: cargo ${{ matrix.command }}
-
+
+  sigma-node-test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+          - command: cargo test --locked --all
+    steps:
+      - uses: webfactory/ssh-agent@v0.9.0
+        with:
+          ssh-private-key: ${{ secrets.SIGMA_NODE_SSH_KEY }}
+      - uses: actions/checkout@v3
+      - uses: dtolnay/rust-toolchain@stable
+      - uses: Swatinem/rust-cache@v2
+      - uses: DeterminateSystems/nix-installer-action@v10
+      - run: nix shell .#sigma-node -c ${{ matrix.command }}
+
   cargo-toml-lint:
     runs-on: ubuntu-latest
     steps:

crates/sigma-stream/src/lib.rs

@@ -3,7 +3,7 @@
 use futures::{Stream, TryStreamExt};
 use reqwest::{ClientBuilder, Url};
 use sigma_types::{Event, Signed};
-use std::marker::PhantomData;
+use std::{fmt::Display, marker::PhantomData};
 use tokio_util::{
     bytes::{self, Buf},
     codec::{Decoder, FramedRead},
@@ -16,6 +16,105 @@ mod tests;
 /// The output type of the event stream - a signed event from the node.
 pub type Output = Signed<Event>;
 
+/// Either the latest height or a specific height.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum Height {
+    /// The latest event height
+    Latest,
+    /// A specific event height
+    Specific(sigma_types::Height),
+}
+
+/// Get the signed events from the node for a specific range of heights.
+pub async fn get_events(
+    url: &Url,
+    start: Height,
+    end: Option<Height>,
+) -> Result<Vec<Output>, std::io::Error> {
+    let client = ClientBuilder::new()
+        .http2_prior_knowledge()
+        .build()
+        .map_err(std::io::Error::other)?;
+
+    let path = match end {
+        Some(h) => format!("get-events/{start}/{h}"),
+        None => format!("get-events/{start}"),
+    };
+    // Create the subscription to the node's batch stream.
+    let url = url
+        .join(&path)
+        .map_err(|e| std::io::Error::other(format!("Invalid URL: {e}")))?;
+
+    let response = client
+        .get(url)
+        .send()
+        .await
+        .map_err(|e| std::io::Error::other(format!("Request failed: {e}")))?;
+
+    // Check if the node returned a bad response.
+    if !response.status().is_success() {
+        let error = format!(
+            "Error connecting to node: {}: {}",
+            response.status(),
+            response
+                .text()
+                .await
+                .unwrap_or_else(|_| "No response body".to_string())
+        );
+        return Err(std::io::Error::other(error));
+    }
+
+    response
+        .json::<Vec<Output>>()
+        .await
+        .map_err(|e| std::io::Error::other(format!("Failed to parse response: {e}")))
+}
+
+/// Get the signed hashes from the node for a specific range of heights.
+pub async fn get_hashes(
+    url: &Url,
+    start: Height,
+    end: Option<Height>,
+) -> Result<Vec<Signed<sigma_types::Hash>>, std::io::Error> {
+    let client = ClientBuilder::new()
+        .http2_prior_knowledge()
+        .build()
+        .map_err(std::io::Error::other)?;
+
+    let path = match end {
+        Some(h) => format!("get-hashes/{start}/{h}"),
+        None => format!("get-hashes/{start}"),
+    };
+    // Create the subscription to the node's batch stream.
+    let url = url
+        .join(&path)
+        .map_err(|e| std::io::Error::other(format!("Invalid URL: {e}")))?;
+
+    let response = client
+        .get(url)
+        .send()
+        .await
+        .map_err(|e| std::io::Error::other(format!("Request failed: {e}")))?;
+
+    // Check if the node returned a bad response.
+    if !response.status().is_success() {
+        let error = format!(
+            "Error connecting to node: {}: {}",
+            response.status(),
+            response
+                .text()
+                .await
+                .unwrap_or_else(|_| "No response body".to_string())
+        );
+        return Err(std::io::Error::other(error));
+    }
+
+    response
+        .json::<Vec<Signed<sigma_types::Hash>>>()
+        .await
+        .map_err(|e| std::io::Error::other(format!("Failed to parse response: {e}")))
+}
+
 /// Create the stream of events from the node endpoint.
 pub async fn stream_events(
     url: &Url,
@@ -29,7 +128,7 @@ pub async fn stream_events(
     // Create the subscription to the node's batch stream.
     let url = url
         .join(&format!("/subscribe-events/{start_event_height}"))
-        .unwrap();
+        .map_err(|e| std::io::Error::other(format!("Invalid URL: {e}")))?;
 
     // Send the request to the node.
     let response = client
@@ -121,3 +220,18 @@ where
         }
     }
 }
+
+impl Display for Height {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match self {
+            Height::Latest => write!(f, "latest"),
+            Height::Specific(h) => write!(f, "{h}"),
+        }
+    }
+}
+
+impl From<sigma_types::Height> for Height {
+    fn from(h: sigma_types::Height) -> Self {
+        Height::Specific(h)
+    }
+}

crates/sigma-stream/tests/common/mod.rs

@@ -1,15 +1,16 @@
-use std::net::TcpStream;
 use std::process::{Child, Command, Stdio};
 use std::thread;
 use std::time::Duration;
 
+use reqwest::ClientBuilder;
+
 pub struct SigmaNode {
     process: Child,
 }
 
 impl SigmaNode {
     /// Start sigma-node if it's available in PATH
-    pub fn start() -> Result<Self, String> {
+    pub async fn start() -> Result<Self, String> {
         // Check if sigma-node is available
         if !Self::is_available() {
             return Err("sigma-node not found in PATH. Please ensure it's installed.".to_string());
@@ -20,8 +21,9 @@ impl SigmaNode {
         cmd.arg("--bind-address").arg("127.0.0.1:3000");
 
         // Add config if it exists
-        if std::path::Path::new("config/test.yaml").exists() {
-            cmd.arg("--config").arg("config/test.yaml");
+        let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../../config/test.yaml");
+        if std::path::Path::new(&path).exists() {
+            cmd.arg("--config").arg(path);
         }
 
         cmd.stdout(Stdio::null()).stderr(Stdio::null());
@@ -31,7 +33,7 @@ impl SigmaNode {
             .map_err(|e| format!("Failed to start sigma-node: {e}"))?;
 
         // Wait for sigma-node to be ready
-        Self::wait_for_ready()?;
+        Self::wait_for_ready().await?;
 
         Ok(SigmaNode { process })
     }
@@ -42,13 +44,20 @@ impl SigmaNode {
     }
 
     /// Wait for sigma-node to be ready
-    fn wait_for_ready() -> Result<(), String> {
+    async fn wait_for_ready() -> Result<(), String> {
         println!("Waiting for sigma-node to start...");
+        let client = ClientBuilder::new()
+            .http2_prior_knowledge()
+            .build()
+            .map_err(|e| format!("Failed to create HTTP client: {e}"))?;
 
         for i in 0..30 {
-            if TcpStream::connect("127.0.0.1:3000").is_ok() {
-                println!("Sigma-node is ready!");
-                return Ok(());
+            let response = client.get("http://127.0.0.1:3000").send().await;
+            if let Ok(response) = response {
+                if response.status().is_success() {
+                    println!("Sigma-node is ready!");
+                    return Ok(());
+                }
             }
             thread::sleep(Duration::from_secs(1));
 
@@ -76,8 +85,8 @@ impl Drop for SigmaNode {
 }
 
 /// Setup function for tests that need sigma-node
-pub fn setup_sigma_node() -> Option<SigmaNode> {
-    match SigmaNode::start() {
+pub async fn setup_sigma_node() -> Option<SigmaNode> {
+    match SigmaNode::start().await {
         Ok(node) => {
             println!("Successfully started sigma-node for tests");
             Some(node)

crates/sigma-stream/tests/gets.rs

@@ -0,0 +1,47 @@
+mod common;
+
+use reqwest::Url;
+use sigma_stream::Height;
+use tokio::time::Duration;
+
+#[tokio::test]
+async fn test_get_events_and_hashes_from_real_endpoint() {
+    // Try to start sigma-node
+    let _node = match common::setup_sigma_node().await {
+        Some(node) => node,
+        None => {
+            println!("Test skipped: sigma-node not available");
+            return;
+        }
+    };
+
+    tokio::time::sleep(Duration::from_secs(3)).await; // Wait for node to start and events to happen
+
+    // Now run the actual test - sigma-node is guaranteed to be running
+    let node_url = Url::parse("http://localhost:3000").expect("Invalid URL");
+    let start_height = 0;
+
+    println!("Connecting to Sigma node at: {node_url}");
+
+    // Get the hashes from the node
+    let hashes = sigma_stream::get_hashes(&node_url, start_height.into(), None)
+        .await
+        .expect("Failed to get hashes from node");
+    assert!(hashes.len() >= 2, "Expected at least two hashes");
+
+    let hashes = sigma_stream::get_hashes(&node_url, start_height.into(), Some(Height::from(2)))
+        .await
+        .expect("Failed to get hashes from node");
+    assert_eq!(hashes.len(), 2, "Expected two hashes");
+
+    // Get the events from the node
+    let events = sigma_stream::get_events(&node_url, start_height.into(), None)
+        .await
+        .expect("Failed to get events from node");
+    assert!(events.len() >= 2, "Expected at least two events");
+
+    let events = sigma_stream::get_events(&node_url, start_height.into(), Some(Height::from(2)))
+        .await
+        .expect("Failed to get events from node");
+    assert_eq!(events.len(), 2, "Expected two events");
+}

crates/sigma-stream/tests/stream_events.rs

@@ -8,7 +8,7 @@ use tokio::time::{Duration, timeout};
 #[tokio::test]
 async fn test_stream_events_from_real_endpoint() {
     // Try to start sigma-node
-    let _node = match common::setup_sigma_node() {
+    let _node = match common::setup_sigma_node().await {
         Some(node) => node,
         None => {
             println!("Test skipped: sigma-node not available");