rsigma_runtime/sources/
mod.rs

1//! Dynamic source resolution for Sigma pipelines.
2//!
3//! This module provides the [`SourceResolver`] trait and a [`DefaultSourceResolver`]
4//! implementation that fetches data from file, command, HTTP, and NATS sources
5//! declared in a pipeline's `sources` section.
6
7pub mod cache;
8pub mod command;
9pub mod extract;
10pub mod file;
11pub mod http;
12pub mod include;
13#[cfg(feature = "nats")]
14pub mod nats;
15pub mod refresh;
16pub mod template;
17
18use std::time::{Duration, Instant};
19
20use rsigma_eval::pipeline::sources::{DynamicSource, ErrorPolicy, SourceType};
21
22/// Maximum size of a source response body (HTTP, command stdout, NATS payload).
23pub const MAX_SOURCE_RESPONSE_BYTES: usize = 10 * 1024 * 1024; // 10 MB
24
25/// Minimum allowed refresh interval to prevent hot CPU loops.
26pub const MIN_REFRESH_INTERVAL: Duration = Duration::from_secs(1);
27
28pub use cache::SourceCache;
29pub use template::TemplateExpander;
30
31/// The result of successfully resolving a dynamic source.
32#[derive(Debug, Clone)]
33pub struct ResolvedValue {
34    /// The resolved data as a YAML value (can be scalar, sequence, or mapping).
35    pub data: serde_json::Value,
36    /// When this value was resolved.
37    pub resolved_at: Instant,
38    /// Whether this value was served from cache rather than freshly fetched.
39    pub from_cache: bool,
40}
41
42/// An error that occurred while resolving a dynamic source.
43#[derive(Debug, Clone)]
44pub struct SourceError {
45    /// The source ID that failed.
46    pub source_id: String,
47    /// What went wrong.
48    pub kind: SourceErrorKind,
49}
50
51impl std::fmt::Display for SourceError {
52    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        write!(f, "source '{}': {}", self.source_id, self.kind)
54    }
55}
56
57impl std::error::Error for SourceError {}
58
59/// The kind of error that occurred during source resolution.
60#[derive(Debug, Clone)]
61pub enum SourceErrorKind {
62    /// Failed to fetch/read the source data.
63    Fetch(String),
64    /// Failed to parse the fetched data into the expected format.
65    Parse(String),
66    /// The `extract` expression failed or returned no data.
67    Extract(String),
68    /// The fetch exceeded the configured timeout.
69    Timeout,
70    /// The response exceeded the maximum allowed size.
71    ResourceLimit(String),
72}
73
74impl std::fmt::Display for SourceErrorKind {
75    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76        match self {
77            Self::Fetch(msg) => write!(f, "fetch failed: {msg}"),
78            Self::Parse(msg) => write!(f, "parse failed: {msg}"),
79            Self::Extract(msg) => write!(f, "extract failed: {msg}"),
80            Self::Timeout => write!(f, "timed out"),
81            Self::ResourceLimit(msg) => write!(f, "resource limit exceeded: {msg}"),
82        }
83    }
84}
85
86/// Trait for resolving dynamic pipeline sources.
87///
88/// Implementations fetch data from external sources (files, commands, HTTP, NATS)
89/// and return it as a JSON value that can be injected into the pipeline.
90#[async_trait::async_trait]
91pub trait SourceResolver: Send + Sync {
92    /// Resolve a single dynamic source, returning the fetched data.
93    async fn resolve(&self, source: &DynamicSource) -> Result<ResolvedValue, SourceError>;
94}
95
96/// Default source resolver that dispatches to file, command, and HTTP resolvers.
97pub struct DefaultSourceResolver {
98    cache: SourceCache,
99}
100
101impl DefaultSourceResolver {
102    /// Create a new resolver with an in-memory cache.
103    pub fn new() -> Self {
104        Self {
105            cache: SourceCache::new(),
106        }
107    }
108
109    /// Create a new resolver with the given cache.
110    pub fn with_cache(cache: SourceCache) -> Self {
111        Self { cache }
112    }
113
114    /// Get a reference to the cache (for inspection/testing).
115    pub fn cache(&self) -> &SourceCache {
116        &self.cache
117    }
118}
119
120impl Default for DefaultSourceResolver {
121    fn default() -> Self {
122        Self::new()
123    }
124}
125
126#[async_trait::async_trait]
127impl SourceResolver for DefaultSourceResolver {
128    async fn resolve(&self, source: &DynamicSource) -> Result<ResolvedValue, SourceError> {
129        let result = match &source.source_type {
130            SourceType::File {
131                path,
132                format,
133                extract,
134            } => file::resolve_file(path, *format, extract.as_ref()).await,
135            SourceType::Command {
136                command,
137                format,
138                extract,
139            } => command::resolve_command(command, *format, extract.as_ref(), source.timeout).await,
140            SourceType::Http {
141                url,
142                method,
143                headers,
144                format,
145                extract,
146            } => {
147                http::resolve_http(
148                    url,
149                    method.as_deref(),
150                    headers,
151                    *format,
152                    extract.as_ref(),
153                    source.timeout,
154                )
155                .await
156            }
157            #[cfg(feature = "nats")]
158            SourceType::Nats {
159                url,
160                subject,
161                format,
162                extract,
163            } => nats::resolve_nats_initial(url, subject, *format, extract.as_ref()).await,
164            #[cfg(not(feature = "nats"))]
165            SourceType::Nats { .. } => {
166                return Err(SourceError {
167                    source_id: source.id.clone(),
168                    kind: SourceErrorKind::Fetch("NATS source requires the 'nats' feature".into()),
169                });
170            }
171        };
172
173        match result {
174            Ok(value) => {
175                self.cache.store(&source.id, &value.data);
176                Ok(value)
177            }
178            Err(mut err) => {
179                err.source_id = source.id.clone();
180                match source.on_error {
181                    ErrorPolicy::UseCached => {
182                        if let Some(cached) = self.cache.get(&source.id) {
183                            tracing::warn!(
184                                source_id = %source.id,
185                                error = %err,
186                                "Source resolution failed, using cached value"
187                            );
188                            Ok(ResolvedValue {
189                                data: cached,
190                                resolved_at: Instant::now(),
191                                from_cache: true,
192                            })
193                        } else {
194                            Err(err)
195                        }
196                    }
197                    ErrorPolicy::UseDefault => {
198                        if let Some(default) = &source.default {
199                            tracing::warn!(
200                                source_id = %source.id,
201                                error = %err,
202                                "Source resolution failed, using default value"
203                            );
204                            let json_default = yaml_value_to_json(default);
205                            Ok(ResolvedValue {
206                                data: json_default,
207                                resolved_at: Instant::now(),
208                                from_cache: false,
209                            })
210                        } else {
211                            Err(err)
212                        }
213                    }
214                    ErrorPolicy::Fail => Err(err),
215                }
216            }
217        }
218    }
219}
220
221/// Resolve all sources in a pipeline, returning a map of source_id -> resolved data.
222///
223/// Applies error policies: `use_cached`, `use_default`, or `fail`.
224/// Required sources with `Fail` policy propagate errors immediately.
225/// Optional sources (required=false) that fail are logged and skipped
226/// with a Null fallback value.
227pub async fn resolve_all(
228    resolver: &dyn SourceResolver,
229    sources: &[DynamicSource],
230) -> Result<std::collections::HashMap<String, serde_json::Value>, SourceError> {
231    resolve_all_with_state(resolver, sources, None).await
232}
233
234/// Like [`resolve_all`] but also updates a [`PipelineState`] with source resolution status.
235pub async fn resolve_all_with_state(
236    resolver: &dyn SourceResolver,
237    sources: &[DynamicSource],
238    mut state: Option<&mut rsigma_eval::pipeline::state::PipelineState>,
239) -> Result<std::collections::HashMap<String, serde_json::Value>, SourceError> {
240    let mut resolved = std::collections::HashMap::new();
241    for source in sources {
242        match resolver.resolve(source).await {
243            Ok(value) => {
244                resolved.insert(source.id.clone(), value.data);
245                if let Some(s) = state.as_deref_mut() {
246                    s.mark_source_resolved(&source.id);
247                }
248            }
249            Err(e) => {
250                if let Some(s) = state.as_deref_mut() {
251                    s.mark_source_failed(&source.id);
252                }
253                if source.required {
254                    return Err(e);
255                }
256                tracing::warn!(
257                    source_id = %source.id,
258                    error = %e,
259                    "Optional source resolution failed, using null"
260                );
261                resolved.insert(source.id.clone(), serde_json::Value::Null);
262            }
263        }
264    }
265    Ok(resolved)
266}
267
268/// Convert a `serde_yaml::Value` to a `serde_json::Value`.
269pub fn yaml_value_to_json(yaml: &serde_yaml::Value) -> serde_json::Value {
270    match yaml {
271        serde_yaml::Value::Null => serde_json::Value::Null,
272        serde_yaml::Value::Bool(b) => serde_json::Value::Bool(*b),
273        serde_yaml::Value::Number(n) => {
274            if let Some(i) = n.as_i64() {
275                serde_json::Value::Number(i.into())
276            } else if let Some(u) = n.as_u64() {
277                serde_json::Value::Number(u.into())
278            } else if let Some(f) = n.as_f64() {
279                serde_json::json!(f)
280            } else {
281                serde_json::Value::Null
282            }
283        }
284        serde_yaml::Value::String(s) => serde_json::Value::String(s.clone()),
285        serde_yaml::Value::Sequence(seq) => {
286            serde_json::Value::Array(seq.iter().map(yaml_value_to_json).collect())
287        }
288        serde_yaml::Value::Mapping(map) => {
289            let obj = map
290                .iter()
291                .map(|(k, v)| {
292                    let key = match k {
293                        serde_yaml::Value::String(s) => s.clone(),
294                        other => format!("{other:?}"),
295                    };
296                    (key, yaml_value_to_json(v))
297                })
298                .collect();
299            serde_json::Value::Object(obj)
300        }
301        serde_yaml::Value::Tagged(tagged) => yaml_value_to_json(&tagged.value),
302    }
303}
rsigma_runtime/sources/mod.rs

rsigma_runtime/sources/
mod.rs
