nidus_integrations/
lib.rs

1#![deny(missing_docs)]
2
3//! Shared primitives for first-party Nidus integration crates.
4//!
5//! This crate deliberately does not define a universal queue or database API.
6//! Broker and data adapters expose their native clients. The types here cover
7//! only stable cross-cutting concerns: envelopes, correlation, redaction-safe
8//! diagnostics, and best-effort lifecycle telemetry.
9
10use std::{collections::BTreeMap, fmt, panic::AssertUnwindSafe, sync::Arc, time::Duration};
11
12use async_trait::async_trait;
13use futures_util::FutureExt;
14use serde::{Deserialize, Serialize, de::DeserializeOwned};
15use thiserror::Error;
16
17/// Default maximum serialized envelope size: one mebibyte.
18pub const DEFAULT_MAX_ENVELOPE_BYTES: usize = 1024 * 1024;
19
20const MAX_NAME_BYTES: usize = 255;
21const MAX_CORRELATION_BYTES: usize = 255;
22const MAX_HEADER_COUNT: usize = 64;
23const MAX_HEADER_NAME_BYTES: usize = 128;
24const MAX_HEADER_VALUE_BYTES: usize = 4096;
25
26/// Result type for shared integration operations.
27pub type Result<T> = std::result::Result<T, IntegrationError>;
28
29/// Validation or serialization error from shared integration primitives.
30#[derive(Debug, Error)]
31pub enum IntegrationError {
32    /// A stable envelope name was empty or too large.
33    #[error("message name must contain 1..={MAX_NAME_BYTES} bytes")]
34    InvalidMessageName,
35    /// A correlation or causation identifier exceeded the safe bound.
36    #[error("correlation identifiers must contain 1..={MAX_CORRELATION_BYTES} bytes")]
37    InvalidCorrelationId,
38    /// A W3C traceparent value was malformed.
39    #[error("traceparent is not a valid W3C trace context value")]
40    InvalidTraceparent,
41    /// An envelope header name was malformed or exceeded its bound.
42    #[error("header names must contain 1..={MAX_HEADER_NAME_BYTES} visible ASCII bytes")]
43    InvalidHeaderName,
44    /// An envelope header value exceeded its bound or contained a control byte.
45    #[error("header values must contain at most {MAX_HEADER_VALUE_BYTES} visible bytes")]
46    InvalidHeaderValue,
47    /// The configured header count bound was exceeded.
48    #[error("an envelope may contain at most {MAX_HEADER_COUNT} headers")]
49    TooManyHeaders,
50    /// The serialized envelope exceeded the configured transport bound.
51    #[error("serialized envelope is {actual} bytes, exceeding the {maximum}-byte limit")]
52    EnvelopeTooLarge {
53        /// Actual serialized or inbound size.
54        actual: usize,
55        /// Configured maximum size.
56        maximum: usize,
57    },
58    /// JSON serialization or deserialization failed.
59    #[error(transparent)]
60    Json(#[from] serde_json::Error),
61}
62
63/// Metadata carried with a message without imposing broker semantics.
64#[derive(Clone, Default, Deserialize, Eq, PartialEq, Serialize)]
65pub struct EnvelopeMetadata {
66    correlation_id: Option<String>,
67    causation_id: Option<String>,
68    traceparent: Option<String>,
69    headers: BTreeMap<String, String>,
70}
71
72impl EnvelopeMetadata {
73    /// Creates empty envelope metadata.
74    pub fn new() -> Self {
75        Self::default()
76    }
77
78    /// Sets a bounded correlation identifier.
79    pub fn correlation_id(mut self, correlation_id: impl Into<String>) -> Result<Self> {
80        self.correlation_id = Some(validate_correlation_id(correlation_id.into())?);
81        Ok(self)
82    }
83
84    /// Sets a bounded causation identifier.
85    pub fn causation_id(mut self, causation_id: impl Into<String>) -> Result<Self> {
86        self.causation_id = Some(validate_correlation_id(causation_id.into())?);
87        Ok(self)
88    }
89
90    /// Sets validated W3C trace context for downstream propagation.
91    pub fn traceparent(mut self, traceparent: impl Into<String>) -> Result<Self> {
92        let traceparent = traceparent.into();
93        if !valid_traceparent(&traceparent) {
94            return Err(IntegrationError::InvalidTraceparent);
95        }
96        self.traceparent = Some(traceparent);
97        Ok(self)
98    }
99
100    /// Adds a bounded transport-neutral header.
101    ///
102    /// Header values remain available to application code but are never shown
103    /// by this type's `Debug` implementation.
104    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Result<Self> {
105        let name = name.into();
106        let value = value.into();
107        validate_header_name(&name)?;
108        validate_header_value(&value)?;
109        if !self.headers.contains_key(&name) && self.headers.len() == MAX_HEADER_COUNT {
110            return Err(IntegrationError::TooManyHeaders);
111        }
112        self.headers.insert(name, value);
113        Ok(self)
114    }
115
116    /// Returns the correlation identifier, when present.
117    pub fn correlation_id_value(&self) -> Option<&str> {
118        self.correlation_id.as_deref()
119    }
120
121    /// Returns the causation identifier, when present.
122    pub fn causation_id_value(&self) -> Option<&str> {
123        self.causation_id.as_deref()
124    }
125
126    /// Returns the propagated W3C traceparent, when present.
127    pub fn traceparent_value(&self) -> Option<&str> {
128        self.traceparent.as_deref()
129    }
130
131    /// Returns application-visible envelope headers.
132    pub fn headers(&self) -> &BTreeMap<String, String> {
133        &self.headers
134    }
135}
136
137impl fmt::Debug for EnvelopeMetadata {
138    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
139        formatter
140            .debug_struct("EnvelopeMetadata")
141            .field("correlation_id", &self.correlation_id)
142            .field("causation_id", &self.causation_id)
143            .field("traceparent", &self.traceparent)
144            .field("header_names", &self.headers.keys().collect::<Vec<_>>())
145            .finish()
146    }
147}
148
149/// Versioned transport-neutral message envelope.
150///
151/// The payload is intentionally omitted from `Debug` output to prevent
152/// accidental secret or PII disclosure. Delivery guarantees remain the
153/// responsibility of each native broker adapter.
154#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
155pub struct MessageEnvelope<T> {
156    id: String,
157    name: String,
158    schema_version: u32,
159    occurred_at_ms: i64,
160    metadata: EnvelopeMetadata,
161    payload: T,
162}
163
164impl<T> MessageEnvelope<T> {
165    /// Creates an envelope with a generated UUID and current UTC timestamp.
166    pub fn new(name: impl Into<String>, payload: T) -> Result<Self> {
167        Self::from_parts(
168            uuid::Uuid::new_v4().to_string(),
169            name,
170            1,
171            time::OffsetDateTime::now_utc().unix_timestamp_nanos() as i64 / 1_000_000,
172            EnvelopeMetadata::new(),
173            payload,
174        )
175    }
176
177    /// Creates an envelope from explicit, validated parts.
178    pub fn from_parts(
179        id: impl Into<String>,
180        name: impl Into<String>,
181        schema_version: u32,
182        occurred_at_ms: i64,
183        metadata: EnvelopeMetadata,
184        payload: T,
185    ) -> Result<Self> {
186        let id = validate_correlation_id(id.into())?;
187        let name = name.into();
188        if name.is_empty() || name.len() > MAX_NAME_BYTES {
189            return Err(IntegrationError::InvalidMessageName);
190        }
191        Ok(Self {
192            id,
193            name,
194            schema_version,
195            occurred_at_ms,
196            metadata,
197            payload,
198        })
199    }
200
201    /// Replaces the envelope metadata.
202    pub fn with_metadata(mut self, metadata: EnvelopeMetadata) -> Self {
203        self.metadata = metadata;
204        self
205    }
206
207    /// Returns the stable message identifier.
208    pub fn id(&self) -> &str {
209        &self.id
210    }
211
212    /// Returns the stable message name.
213    pub fn name(&self) -> &str {
214        &self.name
215    }
216
217    /// Returns the application schema version.
218    pub const fn schema_version(&self) -> u32 {
219        self.schema_version
220    }
221
222    /// Returns the UTC occurrence timestamp in Unix milliseconds.
223    pub const fn occurred_at_ms(&self) -> i64 {
224        self.occurred_at_ms
225    }
226
227    /// Returns the envelope metadata.
228    pub fn metadata(&self) -> &EnvelopeMetadata {
229        &self.metadata
230    }
231
232    /// Returns the typed payload.
233    pub fn payload(&self) -> &T {
234        &self.payload
235    }
236
237    /// Consumes the envelope and returns the typed payload.
238    pub fn into_payload(self) -> T {
239        self.payload
240    }
241}
242
243impl<T> MessageEnvelope<T>
244where
245    T: Serialize,
246{
247    /// Serializes the envelope as bounded JSON.
248    pub fn to_json(&self) -> Result<Vec<u8>> {
249        self.to_json_with_limit(DEFAULT_MAX_ENVELOPE_BYTES)
250    }
251
252    /// Serializes the envelope as JSON with an explicit byte limit.
253    pub fn to_json_with_limit(&self, maximum: usize) -> Result<Vec<u8>> {
254        let encoded = serde_json::to_vec(self)?;
255        if encoded.len() > maximum {
256            return Err(IntegrationError::EnvelopeTooLarge {
257                actual: encoded.len(),
258                maximum,
259            });
260        }
261        Ok(encoded)
262    }
263}
264
265impl<T> MessageEnvelope<T>
266where
267    T: DeserializeOwned,
268{
269    /// Deserializes a bounded JSON envelope.
270    pub fn from_json(encoded: &[u8]) -> Result<Self> {
271        Self::from_json_with_limit(encoded, DEFAULT_MAX_ENVELOPE_BYTES)
272    }
273
274    /// Deserializes JSON after enforcing an explicit inbound byte limit.
275    pub fn from_json_with_limit(encoded: &[u8], maximum: usize) -> Result<Self> {
276        if encoded.len() > maximum {
277            return Err(IntegrationError::EnvelopeTooLarge {
278                actual: encoded.len(),
279                maximum,
280            });
281        }
282        let envelope: Self = serde_json::from_slice(encoded)?;
283        Self::from_parts(
284            envelope.id,
285            envelope.name,
286            envelope.schema_version,
287            envelope.occurred_at_ms,
288            envelope.metadata,
289            envelope.payload,
290        )
291    }
292}
293
294impl<T> fmt::Debug for MessageEnvelope<T> {
295    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
296        formatter
297            .debug_struct("MessageEnvelope")
298            .field("id", &self.id)
299            .field("name", &self.name)
300            .field("schema_version", &self.schema_version)
301            .field("occurred_at_ms", &self.occurred_at_ms)
302            .field("metadata", &self.metadata)
303            .field("payload", &"<redacted>")
304            .finish()
305    }
306}
307
308/// Outcome of a first-party adapter operation.
309#[derive(Clone, Copy, Debug, Eq, PartialEq)]
310pub enum IntegrationStatus {
311    /// The operation completed successfully.
312    Success,
313    /// The operation failed.
314    Failure,
315    /// The operation was cancelled during shutdown.
316    Cancelled,
317}
318
319/// Redaction-safe lifecycle event emitted by an integration adapter.
320#[derive(Clone, Debug, Eq, PartialEq)]
321pub struct IntegrationEvent {
322    adapter: &'static str,
323    operation: &'static str,
324    status: IntegrationStatus,
325    duration: Duration,
326    correlation_id: Option<String>,
327}
328
329impl IntegrationEvent {
330    /// Creates a lifecycle event using stable adapter and operation names.
331    pub fn new(
332        adapter: &'static str,
333        operation: &'static str,
334        status: IntegrationStatus,
335        duration: Duration,
336    ) -> Self {
337        Self {
338            adapter,
339            operation,
340            status,
341            duration,
342            correlation_id: None,
343        }
344    }
345
346    /// Adds a bounded correlation identifier when it is valid.
347    pub fn correlation_id(mut self, correlation_id: impl Into<String>) -> Result<Self> {
348        self.correlation_id = Some(validate_correlation_id(correlation_id.into())?);
349        Ok(self)
350    }
351
352    /// Returns the stable adapter name.
353    pub const fn adapter(&self) -> &'static str {
354        self.adapter
355    }
356
357    /// Returns the stable operation name.
358    pub const fn operation(&self) -> &'static str {
359        self.operation
360    }
361
362    /// Returns the operation outcome.
363    pub const fn status(&self) -> IntegrationStatus {
364        self.status
365    }
366
367    /// Returns the operation duration.
368    pub const fn duration(&self) -> Duration {
369        self.duration
370    }
371
372    /// Returns the correlation identifier, when present.
373    pub fn correlation_id_value(&self) -> Option<&str> {
374        self.correlation_id.as_deref()
375    }
376}
377
378/// Best-effort observer for integration lifecycle operations.
379///
380/// Observers must not include message payloads, raw URLs, credentials, queue
381/// names containing tenant data, or other high-cardinality values in telemetry.
382#[async_trait]
383pub trait IntegrationObserver: Send + Sync + 'static {
384    /// Records one redaction-safe integration event.
385    async fn record(&self, event: &IntegrationEvent);
386}
387
388/// Composite best-effort telemetry used by first-party adapters.
389#[derive(Clone, Default)]
390pub struct IntegrationTelemetry {
391    observers: Arc<Vec<Arc<dyn IntegrationObserver>>>,
392}
393
394impl IntegrationTelemetry {
395    /// Creates telemetry with no observers.
396    pub fn new() -> Self {
397        Self::default()
398    }
399
400    /// Adds a custom integration observer.
401    pub fn with_observer<O>(self, observer: O) -> Self
402    where
403        O: IntegrationObserver,
404    {
405        let mut observers = self.observers.as_ref().clone();
406        observers.push(Arc::new(observer));
407        Self {
408            observers: Arc::new(observers),
409        }
410    }
411
412    /// Adds a redaction-safe `tracing` observer.
413    pub fn tracing(self) -> Self {
414        self.with_observer(TracingIntegrationObserver)
415    }
416
417    /// Adds Nidus Prometheus adapter metrics.
418    #[cfg(feature = "observability")]
419    pub fn observability(
420        self,
421        observer: nidus_observability::ObservabilityAdapterObserver,
422    ) -> Self {
423        self.with_observer(ObservabilityIntegrationObserver(observer))
424    }
425
426    /// Adds Nidus Dashboard adapter timeline capture.
427    #[cfg(feature = "dashboard")]
428    pub fn dashboard(
429        self,
430        collector: nidus_dashboard::DashboardCollector<
431            nidus_dashboard::storage::DashboardStorageHandle,
432        >,
433    ) -> Self {
434        self.with_observer(DashboardIntegrationObserver(collector))
435    }
436
437    /// Returns whether no telemetry observers are configured.
438    pub fn is_empty(&self) -> bool {
439        self.observers.is_empty()
440    }
441
442    /// Records an event to all observers in registration order.
443    pub async fn record(&self, event: &IntegrationEvent) {
444        for observer in self.observers.iter() {
445            if AssertUnwindSafe(observer.record(event))
446                .catch_unwind()
447                .await
448                .is_err()
449            {
450                tracing::warn!(
451                    adapter = event.adapter(),
452                    operation = event.operation(),
453                    "integration telemetry observer panicked and was isolated"
454                );
455            }
456        }
457    }
458}
459
460impl fmt::Debug for IntegrationTelemetry {
461    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
462        formatter
463            .debug_struct("IntegrationTelemetry")
464            .field("observer_count", &self.observers.len())
465            .finish()
466    }
467}
468
469#[derive(Clone, Copy, Debug)]
470struct TracingIntegrationObserver;
471
472#[async_trait]
473impl IntegrationObserver for TracingIntegrationObserver {
474    async fn record(&self, event: &IntegrationEvent) {
475        tracing::info!(
476            adapter = event.adapter(),
477            operation = event.operation(),
478            status = ?event.status(),
479            duration_ms = event.duration().as_millis(),
480            correlation_id = event.correlation_id_value(),
481            "integration operation completed"
482        );
483    }
484}
485
486#[cfg(feature = "observability")]
487#[derive(Clone, Debug)]
488struct ObservabilityIntegrationObserver(nidus_observability::ObservabilityAdapterObserver);
489
490#[cfg(feature = "observability")]
491#[async_trait]
492impl IntegrationObserver for ObservabilityIntegrationObserver {
493    async fn record(&self, event: &IntegrationEvent) {
494        let status = match event.status() {
495            IntegrationStatus::Success => nidus_observability::OperationStatus::Success,
496            IntegrationStatus::Failure | IntegrationStatus::Cancelled => {
497                nidus_observability::OperationStatus::Failure
498            }
499        };
500        self.0
501            .record(event.adapter(), event.operation(), status, event.duration());
502    }
503}
504
505#[cfg(feature = "dashboard")]
506#[derive(Clone, Debug)]
507struct DashboardIntegrationObserver(
508    nidus_dashboard::DashboardCollector<nidus_dashboard::storage::DashboardStorageHandle>,
509);
510
511#[cfg(feature = "dashboard")]
512#[async_trait]
513impl IntegrationObserver for DashboardIntegrationObserver {
514    async fn record(&self, event: &IntegrationEvent) {
515        if let Err(error) = self
516            .0
517            .record_adapter(
518                format!("{}.{}", event.adapter(), event.operation()),
519                event.correlation_id_value(),
520                event.status() == IntegrationStatus::Success,
521                event.duration().as_millis().min(u128::from(u64::MAX)) as u64,
522            )
523            .await
524        {
525            tracing::warn!(error = %error, "dashboard adapter telemetry was dropped");
526        }
527    }
528}
529
530fn validate_correlation_id(value: String) -> Result<String> {
531    if value.is_empty()
532        || value.len() > MAX_CORRELATION_BYTES
533        || value.chars().any(char::is_control)
534    {
535        return Err(IntegrationError::InvalidCorrelationId);
536    }
537    Ok(value)
538}
539
540fn validate_header_name(name: &str) -> Result<()> {
541    if name.is_empty()
542        || name.len() > MAX_HEADER_NAME_BYTES
543        || !name
544            .bytes()
545            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
546    {
547        return Err(IntegrationError::InvalidHeaderName);
548    }
549    Ok(())
550}
551
552fn validate_header_value(value: &str) -> Result<()> {
553    if value.len() > MAX_HEADER_VALUE_BYTES || value.chars().any(char::is_control) {
554        return Err(IntegrationError::InvalidHeaderValue);
555    }
556    Ok(())
557}
558
559fn valid_traceparent(value: &str) -> bool {
560    let bytes = value.as_bytes();
561    if bytes.len() < 55
562        || bytes.len() > MAX_HEADER_VALUE_BYTES
563        || bytes.iter().any(|byte| !(0x21..=0x7e).contains(byte))
564        || bytes[2] != b'-'
565        || bytes[35] != b'-'
566        || bytes[52] != b'-'
567    {
568        return false;
569    }
570
571    let version = &bytes[..2];
572    let trace_id = &bytes[3..35];
573    let parent_id = &bytes[36..52];
574    let flags = &bytes[53..55];
575    let valid_lower_hex = |part: &[u8]| {
576        part.iter()
577            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(byte))
578    };
579    if !valid_lower_hex(version)
580        || version == b"ff"
581        || !valid_lower_hex(trace_id)
582        || trace_id.iter().all(|byte| *byte == b'0')
583        || !valid_lower_hex(parent_id)
584        || parent_id.iter().all(|byte| *byte == b'0')
585        || !valid_lower_hex(flags)
586    {
587        return false;
588    }
589
590    if version == b"00" {
591        bytes.len() == 55
592    } else {
593        bytes.len() == 55 || bytes[55] == b'-'
594    }
595}
nidus_integrations/lib.rs

nidus_integrations/
lib.rs
