arcly_http_core/pipeline/mod.rs
1//! Unified request provenance — one answer to "who is this unit of work,
2//! which trace does it continue, which tenant does it belong to?" shared by
3//! every transport: HTTP requests, WebSocket handshakes, and consumer-mesh
4//! messages.
5//!
6//! ## Why this module exists
7//!
8//! Before it, the three transports drifted: HTTP extracted trace + tenant +
9//! credentials in `assemble_context`, WebSocket handshakes extracted *only*
10//! credentials (no tenant enforcement, orphan-root spans), and the consumer
11//! mesh carried tenant ids as raw strings that never met the
12//! `TenantRegistry` (a suspended tenant's queued events kept processing).
13//! A new identity dimension had to be wired three times — or, in practice,
14//! once, with the other transports silently missing it.
15//!
16//! [`Provenance`] is now the single extraction point. Adding a dimension
17//! here reaches every transport at once; a transport that skips it can be
18//! spotted in review by the absence of one call.
19//!
20//! ## Zero-lock guarantee
21//!
22//! Construction is pure parsing plus frozen-map probes: `traceparent`
23//! parsing, the existing credential pipeline (`auth::extract`), and one
24//! `ArcSwap` snapshot read in the tenant registry. No locks, no I/O beyond
25//! what the credential pipeline already contracted.
26
27use std::sync::Arc;
28
29use http::HeaderMap;
30
31use crate::auth::extract::{extract_auth, extract_claims_ws};
32use crate::core::engine::FrozenDiContainer;
33use crate::observability::propagation::{extract_trace_context, TraceContext};
34use crate::session::Session;
35use crate::transport::InboundMessage;
36use crate::web::context::Claims;
37use crate::web::tenant::{TenantConfig, TenantRegistry};
38
39/// Who/where/why for one unit of work — identical shape across transports.
40pub struct Provenance {
41 /// W3C trace identity: continued from the caller/producer, or a fresh root.
42 pub trace: TraceContext,
43 /// Resolved + validated tenant (`None` = unknown without fallback, or
44 /// suspended — callers must treat suspended as a hard cut-off).
45 pub tenant: Option<Arc<TenantConfig>>,
46 /// Decoded principal claims (JWT / signed cookie), when presented.
47 pub claims: Option<Arc<Claims>>,
48 /// Server-side session, when the session pipeline matched a cookie.
49 pub session: Option<Arc<Session>>,
50}
51
52impl Provenance {
53 /// HTTP requests: headers carry everything.
54 ///
55 /// This is the ONE place trace + tenant + credentials meet a header map.
56 pub async fn from_headers(headers: &HeaderMap, container: &FrozenDiContainer) -> Self {
57 let trace = extract_trace_context(headers);
58 let auth = extract_auth(headers, container).await;
59 let tenant = container
60 .try_get::<TenantRegistry>()
61 .and_then(|tr| tr.resolve(headers));
62 Self {
63 trace,
64 tenant,
65 claims: auth.claims,
66 session: auth.session,
67 }
68 }
69
70 /// WebSocket handshakes: headers, plus `?access_token=<jwt>` from the URL.
71 ///
72 /// Identical to [`Provenance::from_headers`] except for that one extra
73 /// credential source — browsers cannot set headers on `new WebSocket(url)`,
74 /// so without it no browser can authenticate a handshake at all. The query
75 /// token is only consulted when no header/cookie credential is present, and
76 /// it decodes through the same access-JWT path, so the claims a gateway sees
77 /// are the same either way. See [`extract_claims_ws`].
78 ///
79 /// No server-side session is loaded: a WS connection authenticates once at
80 /// upgrade and holds its identity for the life of the socket.
81 pub fn from_ws_handshake(
82 headers: &HeaderMap,
83 query: Option<&str>,
84 container: &FrozenDiContainer,
85 ) -> Self {
86 let trace = extract_trace_context(headers);
87 let claims = extract_claims_ws(headers, query, container);
88 let tenant = container
89 .try_get::<TenantRegistry>()
90 .and_then(|tr| tr.resolve(headers));
91 Self {
92 trace,
93 tenant,
94 claims,
95 session: None,
96 }
97 }
98
99 /// Consumer-mesh messages: the producing request's identity rides the
100 /// envelope. The tenant id is validated against the SAME registry as
101 /// HTTP — a suspended tenant's queued events resolve to `None` and stop
102 /// being processed — and the producer's `traceparent` continues the
103 /// distributed trace instead of starting an orphan root.
104 ///
105 /// Messages authenticate at the transport (broker credentials), not per
106 /// event, so `claims`/`session` are absent by design.
107 pub fn from_message(msg: &InboundMessage, container: &FrozenDiContainer) -> Self {
108 let trace = msg
109 .traceparent
110 .as_deref()
111 .and_then(TraceContext::from_traceparent)
112 .unwrap_or_else(TraceContext::new_root);
113 let tenant = msg.tenant.as_deref().and_then(|id| {
114 container
115 .try_get::<TenantRegistry>()
116 .and_then(|tr| tr.resolve_by_id(id))
117 });
118 Self {
119 trace,
120 tenant,
121 claims: None,
122 session: None,
123 }
124 }
125
126 /// `traceparent` for stamping outbound hops (HTTP calls, outbox rows,
127 /// queue envelopes).
128 pub fn traceparent(&self) -> String {
129 self.trace.to_traceparent()
130 }
131}
132
133#[cfg(test)]
134mod tests {
135 use super::*;
136 use crate::core::engine::DiContainerBuilder;
137 use crate::web::tenant::{TenantId, TenantStrategy};
138
139 fn container_with_registry() -> Arc<FrozenDiContainer> {
140 let registry = TenantRegistry::new(
141 TenantStrategy::header("x-tenant-id"),
142 vec![TenantConfig {
143 id: TenantId::new("acme"),
144 display_name: "Acme".into(),
145 datasource: "acme".into(),
146 }],
147 None,
148 );
149 registry.suspend(&TenantId::new("globex")); // suspended, never known
150 let mut b = DiContainerBuilder::new();
151 b.register(registry);
152 b.freeze()
153 }
154
155 fn msg(tenant: Option<&str>, traceparent: Option<&str>) -> InboundMessage {
156 InboundMessage {
157 topic: "t".into(),
158 payload: serde_json::Value::Null,
159 idempotency_key: "k".into(),
160 tenant: tenant.map(str::to_owned),
161 traceparent: traceparent.map(str::to_owned),
162 }
163 }
164
165 #[test]
166 fn message_provenance_validates_tenant_against_registry() {
167 let c = container_with_registry();
168
169 // Known tenant resolves to the same config HTTP would see.
170 let p = Provenance::from_message(&msg(Some("acme"), None), &c);
171 assert_eq!(p.tenant.as_deref().map(|t| t.id.as_str()), Some("acme"));
172
173 // Unknown and suspended ids resolve to None — handlers never see them.
174 assert!(Provenance::from_message(&msg(Some("nope"), None), &c)
175 .tenant
176 .is_none());
177 assert!(Provenance::from_message(&msg(Some("globex"), None), &c)
178 .tenant
179 .is_none());
180 // No tenant on the envelope: simply absent.
181 assert!(Provenance::from_message(&msg(None, None), &c)
182 .tenant
183 .is_none());
184 }
185
186 #[test]
187 fn message_provenance_continues_producer_trace() {
188 let c = container_with_registry();
189 let carried = "00-0123456789abcdef0123456789abcdef-00f067aa0ba902b7-01";
190
191 let p = Provenance::from_message(&msg(None, Some(carried)), &c);
192 let hex = crate::observability::lean_telemetry::hex_encode(&p.trace.trace_id);
193 assert_eq!(
194 hex, "0123456789abcdef0123456789abcdef",
195 "trace id continues"
196 );
197 // Producer's span is this hop's parent; a fresh consumer span is minted.
198 let parent = crate::observability::lean_telemetry::hex_encode(&p.trace.parent_span_id);
199 assert_eq!(parent, "00f067aa0ba902b7");
200
201 // No traceparent → fresh root (all-zero parent).
202 let root = Provenance::from_message(&msg(None, None), &c);
203 assert_eq!(root.trace.parent_span_id, [0u8; 8]);
204 }
205}