feat: dynamic pipeline source resolution engine (Phase 2-3)

crates/rsigma-cli/Cargo.toml

@@ -11,7 +11,7 @@ homepage.workspace = true
 
 [features]
 default = ["daemon"]
-daemon = ["rsigma-runtime", "tokio", "axum", "prometheus", "notify", "rusqlite"]
+daemon = ["rsigma-runtime", "tokio", "axum", "async-trait", "prometheus", "notify", "rusqlite"]
 daemon-nats = ["daemon", "rsigma-runtime/nats", "async-nats", "tokio-stream", "time"]
 daemon-otlp = ["daemon", "rsigma-runtime/otlp", "prost", "tonic", "flate2", "tokio-stream"]
 logfmt = ["rsigma-runtime/logfmt"]
@@ -43,6 +43,7 @@ chrono = { version = "0.4", default-features = false, features = ["std", "now"]
 # daemon mode dependencies
 tokio = { version = "1", features = ["full"], optional = true }
 axum = { version = "0.8", features = ["json"], optional = true }
+async-trait = { version = "0.1", optional = true }
 prometheus = { version = "0.14", default-features = false, optional = true }
 notify = { version = "8.2", optional = true }
 rusqlite = { version = "0.39", features = ["bundled"], optional = true }

crates/rsigma-cli/src/commands/mod.rs

@@ -3,11 +3,13 @@ mod eval;
 mod fields;
 mod lint;
 mod parse;
+mod resolve;
 mod validate;
 
 pub(crate) use convert::{cmd_convert, cmd_list_formats, cmd_list_targets};
 pub(crate) use eval::cmd_eval;
 pub(crate) use fields::cmd_fields;
 pub(crate) use lint::cmd_lint;
 pub(crate) use parse::{cmd_condition, cmd_parse, cmd_stdin};
+pub(crate) use resolve::cmd_resolve;
 pub(crate) use validate::cmd_validate;

crates/rsigma-cli/src/commands/resolve.rs

@@ -0,0 +1,104 @@
+//! CLI `resolve` command: test dynamic source resolution offline.
+
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use rsigma_eval::parse_pipeline_file;
+use rsigma_runtime::DefaultSourceResolver;
+use rsigma_runtime::sources::SourceResolver;
+
+pub fn cmd_resolve(pipeline_paths: Vec<PathBuf>, source_filter: Option<String>, pretty: bool) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap_or_else(|e| {
+            eprintln!("Failed to start async runtime: {e}");
+            std::process::exit(crate::exit_code::CONFIG_ERROR);
+        });
+
+    rt.block_on(async { resolve_async(pipeline_paths, source_filter, pretty).await });
+}
+
+async fn resolve_async(pipeline_paths: Vec<PathBuf>, source_filter: Option<String>, pretty: bool) {
+    let mut all_sources = Vec::new();
+
+    for path in &pipeline_paths {
+        let pipeline = match parse_pipeline_file(path) {
+            Ok(p) => p,
+            Err(e) => {
+                eprintln!("Error reading pipeline {}: {e}", path.display());
+                std::process::exit(crate::exit_code::RULE_ERROR);
+            }
+        };
+
+        if !pipeline.is_dynamic() {
+            eprintln!(
+                "Pipeline '{}' has no dynamic sources, skipping.",
+                pipeline.name
+            );
+            continue;
+        }
+
+        for source in &pipeline.sources {
+            if let Some(ref filter) = source_filter
+                && source.id != *filter
+            {
+                continue;
+            }
+            all_sources.push((pipeline.name.clone(), source.clone()));
+        }
+    }
+
+    if all_sources.is_empty() {
+        if source_filter.is_some() {
+            eprintln!("No sources matched the filter.");
+        } else {
+            eprintln!("No dynamic sources found in the provided pipelines.");
+        }
+        std::process::exit(crate::exit_code::RULE_ERROR);
+    }
+
+    let resolver = Arc::new(DefaultSourceResolver::new());
+    let mut results = Vec::new();
+    let mut had_error = false;
+
+    for (pipeline_name, source) in &all_sources {
+        let source_id = source.id.clone();
+        match resolver.resolve(source).await {
+            Ok(value) => {
+                results.push(serde_json::json!({
+                    "pipeline": pipeline_name,
+                    "source_id": source_id,
+                    "status": "ok",
+                    "data": value.data,
+                }));
+            }
+            Err(e) => {
+                had_error = true;
+                results.push(serde_json::json!({
+                    "pipeline": pipeline_name,
+                    "source_id": source_id,
+                    "status": "error",
+                    "error": e.to_string(),
+                }));
+            }
+        }
+    }
+
+    let output = if results.len() == 1 {
+        results.into_iter().next().unwrap()
+    } else {
+        serde_json::Value::Array(results)
+    };
+
+    let json_str = if pretty {
+        serde_json::to_string_pretty(&output).unwrap()
+    } else {
+        serde_json::to_string(&output).unwrap()
+    };
+    println!("{json_str}");
+
+    if had_error {
+        std::process::exit(1);
+    }
+}