substrate_trace/
lib.rs

1//! # substrate-trace
2//!
3//! Concrete [`TracePort`] adapters. The trait itself is defined in
4//! `substrate-core` so the application layer can emit events without
5//! depending on any adapter crate.
6//!
7//! ## Adapters
8//!
9//! | Type | Purpose |
10//! |------|---------|
11//! | [`NoopTrace`] | Silently discards all events (useful as a default). |
12//! | [`RecordingTrace`] | Stores events in memory for test assertions. |
13//! | [`MultiTrace`] | Fans a single event stream out to N [`TracePort`]s. |
14//! | [`AgilePlusTrace`] | POSTs events to the AgilePlus API. |
15//! | [`TraceraTrace`] | POSTs events to the Tracera API. |
16#![forbid(unsafe_code)]
17#![warn(missing_docs)]
18
19use std::sync::{Arc, Mutex};
20
21use substrate_core::trace::{TaskCompleted, TaskFailed, TaskRegistered, TracePort};
22
23// ---------------------------------------------------------------------------
24// NoopTrace
25// ---------------------------------------------------------------------------
26
27/// A [`TracePort`] that silently discards every event.
28///
29/// Useful as a default when no trace backend is configured.
30#[derive(Debug, Default, Clone)]
31pub struct NoopTrace;
32
33impl TracePort for NoopTrace {
34    fn task_registered(&self, _event: TaskRegistered) {}
35    fn task_completed(&self, _event: TaskCompleted) {}
36    fn task_failed(&self, _event: TaskFailed) {}
37}
38
39// ---------------------------------------------------------------------------
40// TraceEvent (union for RecordingTrace)
41// ---------------------------------------------------------------------------
42
43/// A discriminated union of the three trace event kinds, stored by
44/// [`RecordingTrace`].
45#[derive(Debug, Clone)]
46pub enum TraceEvent {
47    /// A [`TaskRegistered`] event.
48    Registered(TaskRegistered),
49    /// A [`TaskCompleted`] event.
50    Completed(TaskCompleted),
51    /// A [`TaskFailed`] event.
52    Failed(TaskFailed),
53}
54
55// ---------------------------------------------------------------------------
56// RecordingTrace
57// ---------------------------------------------------------------------------
58
59/// An in-memory [`TracePort`] that records every event for later inspection.
60///
61/// Thread-safe via `Arc<Mutex<…>>` so it can be cloned and shared across
62/// threads in test assertions.
63#[derive(Debug, Clone, Default)]
64pub struct RecordingTrace {
65    events: Arc<Mutex<Vec<TraceEvent>>>,
66}
67
68impl RecordingTrace {
69    /// Create a new, empty recording trace.
70    pub fn new() -> Self {
71        RecordingTrace::default()
72    }
73
74    /// Return a snapshot of all recorded events in arrival order.
75    pub fn events(&self) -> Vec<TraceEvent> {
76        self.events
77            .lock()
78            .expect("RecordingTrace lock poisoned")
79            .clone()
80    }
81
82    /// Return the number of recorded events.
83    pub fn len(&self) -> usize {
84        self.events
85            .lock()
86            .expect("RecordingTrace lock poisoned")
87            .len()
88    }
89
90    /// Returns true if no events have been recorded.
91    pub fn is_empty(&self) -> bool {
92        self.len() == 0
93    }
94}
95
96impl TracePort for RecordingTrace {
97    fn task_registered(&self, event: TaskRegistered) {
98        self.events
99            .lock()
100            .expect("RecordingTrace lock poisoned")
101            .push(TraceEvent::Registered(event));
102    }
103
104    fn task_completed(&self, event: TaskCompleted) {
105        self.events
106            .lock()
107            .expect("RecordingTrace lock poisoned")
108            .push(TraceEvent::Completed(event));
109    }
110
111    fn task_failed(&self, event: TaskFailed) {
112        self.events
113            .lock()
114            .expect("RecordingTrace lock poisoned")
115            .push(TraceEvent::Failed(event));
116    }
117}
118
119// ---------------------------------------------------------------------------
120// MultiTrace
121// ---------------------------------------------------------------------------
122
123/// A [`TracePort`] that fans every event out to N downstream [`TracePort`]s.
124///
125/// Useful for shipping to both AgilePlus and Tracera simultaneously, or for
126/// augmenting a production backend with a [`RecordingTrace`] in tests.
127pub struct MultiTrace {
128    sinks: Vec<Arc<dyn TracePort>>,
129}
130
131impl std::fmt::Debug for MultiTrace {
132    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
133        f.debug_struct("MultiTrace")
134            .field("sink_count", &self.sinks.len())
135            .finish()
136    }
137}
138
139impl MultiTrace {
140    /// Create a fan-out trace with the given sinks.
141    pub fn new(sinks: Vec<Arc<dyn TracePort>>) -> Self {
142        MultiTrace { sinks }
143    }
144
145    /// Create an empty fan-out (equivalent to [`NoopTrace`]; add sinks with
146    /// [`MultiTrace::with_sink`]).
147    pub fn empty() -> Self {
148        MultiTrace { sinks: vec![] }
149    }
150
151    /// Append a sink and return `self` for chained construction.
152    pub fn with_sink(mut self, sink: Arc<dyn TracePort>) -> Self {
153        self.sinks.push(sink);
154        self
155    }
156}
157
158impl TracePort for MultiTrace {
159    fn task_registered(&self, event: TaskRegistered) {
160        for sink in &self.sinks {
161            sink.task_registered(event.clone());
162        }
163    }
164
165    fn task_completed(&self, event: TaskCompleted) {
166        for sink in &self.sinks {
167            sink.task_completed(event.clone());
168        }
169    }
170
171    fn task_failed(&self, event: TaskFailed) {
172        for sink in &self.sinks {
173            sink.task_failed(event.clone());
174        }
175    }
176}
177
178// ---------------------------------------------------------------------------
179// AgilePlusTrace
180// ---------------------------------------------------------------------------
181
182/// Payload sent to the AgilePlus API for a registered task.
183#[derive(Debug, serde::Serialize)]
184struct AgilePlusRegistered<'a> {
185    task_id: &'a str,
186    #[serde(skip_serializing_if = "Option::is_none")]
187    requirement_id: Option<&'a str>,
188    #[serde(skip_serializing_if = "Option::is_none")]
189    epic_id: Option<&'a str>,
190}
191
192/// Payload sent to the AgilePlus API for a completed task.
193#[derive(Debug, serde::Serialize)]
194struct AgilePlusCompleted<'a> {
195    task_id: &'a str,
196    pr_urls: &'a [String],
197    #[serde(skip_serializing_if = "Option::is_none")]
198    requirement_id: Option<&'a str>,
199}
200
201/// Payload sent to the AgilePlus API for a failed task.
202#[derive(Debug, serde::Serialize)]
203struct AgilePlusFailed<'a> {
204    task_id: &'a str,
205    error: &'a str,
206    #[serde(skip_serializing_if = "Option::is_none")]
207    requirement_id: Option<&'a str>,
208}
209
210/// A [`TracePort`] that POSTs events to the AgilePlus API.
211///
212/// The endpoint base URL is read from the `AGILEPLUS_ENDPOINT` env var at
213/// construction time. HTTP errors are silently swallowed (trace shipping
214/// must never fail a dispatch).
215#[derive(Debug, Clone)]
216pub struct AgilePlusTrace {
217    endpoint: String,
218    client: reqwest::Client,
219    rt: Arc<tokio::runtime::Handle>,
220}
221
222impl AgilePlusTrace {
223    /// Construct from the `AGILEPLUS_ENDPOINT` env var.
224    ///
225    /// Panics if called outside a Tokio runtime context (uses
226    /// [`tokio::runtime::Handle::current()`]).
227    pub fn from_env() -> Self {
228        let endpoint = std::env::var("AGILEPLUS_ENDPOINT")
229            .unwrap_or_else(|_| "http://localhost:4000".to_string());
230        AgilePlusTrace {
231            endpoint,
232            client: reqwest::Client::new(),
233            rt: Arc::new(tokio::runtime::Handle::current()),
234        }
235    }
236
237    /// Construct with an explicit endpoint URL.
238    pub fn with_endpoint(endpoint: impl Into<String>) -> Self {
239        AgilePlusTrace {
240            endpoint: endpoint.into(),
241            client: reqwest::Client::new(),
242            rt: Arc::new(tokio::runtime::Handle::current()),
243        }
244    }
245}
246
247impl TracePort for AgilePlusTrace {
248    fn task_registered(&self, event: TaskRegistered) {
249        let body = AgilePlusRegistered {
250            task_id: &event.task_id,
251            requirement_id: event.requirement_id.as_deref(),
252            epic_id: event.epic_id.as_deref(),
253        };
254        // We must own the data before spawning; serialize to JSON string.
255        if let Ok(json) = serde_json::to_string(&body) {
256            let url = format!("{}/v1/tasks/registered", self.endpoint);
257            let client = self.client.clone();
258            self.rt.spawn(async move {
259                let _ = client
260                    .post(&url)
261                    .header("content-type", "application/json")
262                    .body(json)
263                    .send()
264                    .await;
265            });
266        }
267    }
268
269    fn task_completed(&self, event: TaskCompleted) {
270        if let Ok(json) = serde_json::to_string(&AgilePlusCompleted {
271            task_id: &event.task_id,
272            pr_urls: &event.pr_urls,
273            requirement_id: event.requirement_id.as_deref(),
274        }) {
275            let url = format!("{}/v1/tasks/completed", self.endpoint);
276            let client = self.client.clone();
277            self.rt.spawn(async move {
278                let _ = client
279                    .post(&url)
280                    .header("content-type", "application/json")
281                    .body(json)
282                    .send()
283                    .await;
284            });
285        }
286    }
287
288    fn task_failed(&self, event: TaskFailed) {
289        if let Ok(json) = serde_json::to_string(&AgilePlusFailed {
290            task_id: &event.task_id,
291            error: &event.error,
292            requirement_id: event.requirement_id.as_deref(),
293        }) {
294            let url = format!("{}/v1/tasks/failed", self.endpoint);
295            let client = self.client.clone();
296            self.rt.spawn(async move {
297                let _ = client
298                    .post(&url)
299                    .header("content-type", "application/json")
300                    .body(json)
301                    .send()
302                    .await;
303            });
304        }
305    }
306}
307
308// ---------------------------------------------------------------------------
309// TraceraTrace
310// ---------------------------------------------------------------------------
311
312/// A [`TracePort`] that POSTs events to the Tracera API.
313///
314/// The endpoint base URL is read from the `TRACERA_ENDPOINT` env var at
315/// construction time. HTTP errors are silently swallowed.
316#[derive(Debug, Clone)]
317pub struct TraceraTrace {
318    endpoint: String,
319    client: reqwest::Client,
320    rt: Arc<tokio::runtime::Handle>,
321}
322
323impl TraceraTrace {
324    /// Construct from the `TRACERA_ENDPOINT` env var.
325    pub fn from_env() -> Self {
326        let endpoint = std::env::var("TRACERA_ENDPOINT")
327            .unwrap_or_else(|_| "http://localhost:5000".to_string());
328        TraceraTrace {
329            endpoint,
330            client: reqwest::Client::new(),
331            rt: Arc::new(tokio::runtime::Handle::current()),
332        }
333    }
334
335    /// Construct with an explicit endpoint URL.
336    pub fn with_endpoint(endpoint: impl Into<String>) -> Self {
337        TraceraTrace {
338            endpoint: endpoint.into(),
339            client: reqwest::Client::new(),
340            rt: Arc::new(tokio::runtime::Handle::current()),
341        }
342    }
343}
344
345impl TracePort for TraceraTrace {
346    fn task_registered(&self, event: TaskRegistered) {
347        if let Ok(json) = serde_json::to_string(&serde_json::json!({
348            "task_id": event.task_id,
349            "requirement_id": event.requirement_id,
350            "epic_id": event.epic_id,
351        })) {
352            let url = format!("{}/api/tasks/registered", self.endpoint);
353            let client = self.client.clone();
354            self.rt.spawn(async move {
355                let _ = client
356                    .post(&url)
357                    .header("content-type", "application/json")
358                    .body(json)
359                    .send()
360                    .await;
361            });
362        }
363    }
364
365    fn task_completed(&self, event: TaskCompleted) {
366        if let Ok(json) = serde_json::to_string(&serde_json::json!({
367            "task_id": event.task_id,
368            "pr_urls": event.pr_urls,
369            "requirement_id": event.requirement_id,
370        })) {
371            let url = format!("{}/api/tasks/completed", self.endpoint);
372            let client = self.client.clone();
373            self.rt.spawn(async move {
374                let _ = client
375                    .post(&url)
376                    .header("content-type", "application/json")
377                    .body(json)
378                    .send()
379                    .await;
380            });
381        }
382    }
383
384    fn task_failed(&self, event: TaskFailed) {
385        if let Ok(json) = serde_json::to_string(&serde_json::json!({
386            "task_id": event.task_id,
387            "error": event.error,
388            "requirement_id": event.requirement_id,
389        })) {
390            let url = format!("{}/api/tasks/failed", self.endpoint);
391            let client = self.client.clone();
392            self.rt.spawn(async move {
393                let _ = client
394                    .post(&url)
395                    .header("content-type", "application/json")
396                    .body(json)
397                    .send()
398                    .await;
399            });
400        }
401    }
402}
403
404// ---------------------------------------------------------------------------
405// Tests
406// ---------------------------------------------------------------------------
407
408#[cfg(test)]
409mod tests {
410    use super::*;
411
412    // ── NoopTrace ────────────────────────────────────────────────────────────
413
414    #[test]
415    fn noop_trace_is_inert() {
416        let t = NoopTrace;
417        // Should not panic.
418        t.task_registered(TaskRegistered {
419            task_id: "t1".into(),
420            requirement_id: None,
421            epic_id: None,
422        });
423        t.task_completed(TaskCompleted {
424            task_id: "t1".into(),
425            pr_urls: vec![],
426            requirement_id: None,
427        });
428        t.task_failed(TaskFailed {
429            task_id: "t1".into(),
430            error: "oops".into(),
431            requirement_id: None,
432        });
433    }
434
435    // ── RecordingTrace ────────────────────────────────────────────────────────
436
437    #[test]
438    fn recording_trace_starts_empty() {
439        let r = RecordingTrace::new();
440        assert!(r.is_empty());
441        assert_eq!(r.len(), 0);
442    }
443
444    #[test]
445    fn recording_trace_captures_lifecycle() {
446        let r = RecordingTrace::new();
447
448        r.task_registered(TaskRegistered {
449            task_id: "task-1".into(),
450            requirement_id: Some("FR-42".into()),
451            epic_id: Some("E-1".into()),
452        });
453        assert_eq!(r.len(), 1);
454        assert!(matches!(&r.events()[0], TraceEvent::Registered(e) if e.task_id == "task-1"));
455
456        r.task_completed(TaskCompleted {
457            task_id: "task-1".into(),
458            pr_urls: vec!["https://github.com/foo/bar/pull/1".into()],
459            requirement_id: Some("FR-42".into()),
460        });
461        assert_eq!(r.len(), 2);
462        assert!(matches!(&r.events()[1], TraceEvent::Completed(e) if e.task_id == "task-1"));
463    }
464
465    #[test]
466    fn recording_trace_captures_failure() {
467        let r = RecordingTrace::new();
468        r.task_registered(TaskRegistered {
469            task_id: "task-2".into(),
470            requirement_id: None,
471            epic_id: None,
472        });
473        r.task_failed(TaskFailed {
474            task_id: "task-2".into(),
475            error: "engine timeout".into(),
476            requirement_id: None,
477        });
478        assert_eq!(r.len(), 2);
479        assert!(matches!(&r.events()[1], TraceEvent::Failed(e) if e.error == "engine timeout"));
480    }
481
482    // ── MultiTrace ────────────────────────────────────────────────────────────
483
484    #[test]
485    fn multi_trace_fans_to_n_consumers() {
486        let r1 = Arc::new(RecordingTrace::new());
487        let r2 = Arc::new(RecordingTrace::new());
488        let r3 = Arc::new(RecordingTrace::new());
489
490        let multi = MultiTrace::new(vec![
491            r1.clone() as Arc<dyn TracePort>,
492            r2.clone() as Arc<dyn TracePort>,
493            r3.clone() as Arc<dyn TracePort>,
494        ]);
495
496        multi.task_registered(TaskRegistered {
497            task_id: "t".into(),
498            requirement_id: None,
499            epic_id: None,
500        });
501        multi.task_completed(TaskCompleted {
502            task_id: "t".into(),
503            pr_urls: vec![],
504            requirement_id: None,
505        });
506
507        for r in [&r1, &r2, &r3] {
508            assert_eq!(r.len(), 2, "each sink must receive both events");
509        }
510    }
511
512    #[test]
513    fn multi_trace_empty_is_noop() {
514        let multi = MultiTrace::empty();
515        // Must not panic.
516        multi.task_registered(TaskRegistered {
517            task_id: "t".into(),
518            requirement_id: None,
519            epic_id: None,
520        });
521    }
522
523    #[test]
524    fn multi_trace_with_sink_builder() {
525        let r = Arc::new(RecordingTrace::new());
526        let multi = MultiTrace::empty().with_sink(r.clone() as Arc<dyn TracePort>);
527        multi.task_failed(TaskFailed {
528            task_id: "t".into(),
529            error: "x".into(),
530            requirement_id: None,
531        });
532        assert_eq!(r.len(), 1);
533    }
534
535    // ── Dispatch lifecycle (trace integration) ────────────────────────────────
536
537    #[test]
538    fn dispatch_emits_registered_then_completed() {
539        // Simulate what DispatchService does: emit Registered, then Completed.
540        let r = RecordingTrace::new();
541        let task_id = "lifecycle-1".to_string();
542
543        r.task_registered(TaskRegistered {
544            task_id: task_id.clone(),
545            requirement_id: Some("FR-1".into()),
546            epic_id: None,
547        });
548        r.task_completed(TaskCompleted {
549            task_id: task_id.clone(),
550            pr_urls: vec!["https://github.com/foo/bar/pull/42".into()],
551            requirement_id: Some("FR-1".into()),
552        });
553
554        let events = r.events();
555        assert_eq!(events.len(), 2);
556        // First event must be Registered.
557        assert!(
558            matches!(&events[0], TraceEvent::Registered(e) if e.task_id == task_id),
559            "first event must be Registered"
560        );
561        // Second event must be Completed.
562        assert!(
563            matches!(&events[1], TraceEvent::Completed(e) if e.pr_urls.len() == 1),
564            "second event must be Completed with pr_url"
565        );
566    }
567
568    #[test]
569    fn dispatch_emits_registered_then_failed() {
570        let r = RecordingTrace::new();
571        let task_id = "lifecycle-2".to_string();
572
573        r.task_registered(TaskRegistered {
574            task_id: task_id.clone(),
575            requirement_id: None,
576            epic_id: None,
577        });
578        r.task_failed(TaskFailed {
579            task_id: task_id.clone(),
580            error: "engine exited non-zero".into(),
581            requirement_id: None,
582        });
583
584        let events = r.events();
585        assert_eq!(events.len(), 2);
586        assert!(matches!(&events[0], TraceEvent::Registered(_)));
587        assert!(matches!(&events[1], TraceEvent::Failed(e) if e.error.contains("engine")));
588    }
589}
substrate_trace/lib.rs

substrate_trace/
lib.rs
