arcly_http/pipeline/

mod.rs

//! Unified request provenance — one answer to "who is this unit of work,
//! which trace does it continue, which tenant does it belong to?" shared by
//! every transport: HTTP requests, WebSocket handshakes, and consumer-mesh
//! messages.
//!
//! ## Why this module exists
//!
//! Before it, the three transports drifted: HTTP extracted trace + tenant +
//! credentials in `assemble_context`, WebSocket handshakes extracted *only*
//! credentials (no tenant enforcement, orphan-root spans), and the consumer
//! mesh carried tenant ids as raw strings that never met the
//! `TenantRegistry` (a suspended tenant's queued events kept processing).
//! A new identity dimension had to be wired three times — or, in practice,
//! once, with the other transports silently missing it.
//!
//! [`Provenance`] is now the single extraction point. Adding a dimension
//! here reaches every transport at once; a transport that skips it can be
//! spotted in review by the absence of one call.
//!
//! ## Zero-lock guarantee
//!
//! Construction is pure parsing plus frozen-map probes: `traceparent`
//! parsing, the existing credential pipeline (`auth::extract`), and one
//! `ArcSwap` snapshot read in the tenant registry. No locks, no I/O beyond
//! what the credential pipeline already contracted.

use std::sync::Arc;

use axum::http::HeaderMap;

use crate::auth::extract::extract_auth;
use crate::core::engine::FrozenDiContainer;
use crate::messaging::InboundMessage;
use crate::observability::propagation::{extract_trace_context, TraceContext};
use crate::session::Session;
use crate::web::context::Claims;
use crate::web::tenant::{TenantConfig, TenantRegistry};

/// Who/where/why for one unit of work — identical shape across transports.
pub struct Provenance {
    /// W3C trace identity: continued from the caller/producer, or a fresh root.
    pub trace: TraceContext,
    /// Resolved + validated tenant (`None` = unknown without fallback, or
    /// suspended — callers must treat suspended as a hard cut-off).
    pub tenant: Option<Arc<TenantConfig>>,
    /// Decoded principal claims (JWT / signed cookie), when presented.
    pub claims: Option<Arc<Claims>>,
    /// Server-side session, when the session pipeline matched a cookie.
    pub session: Option<Arc<Session>>,
}

impl Provenance {
    /// HTTP requests and WebSocket handshakes: headers carry everything.
    ///
    /// This is the ONE place trace + tenant + credentials meet a header map.
    pub async fn from_headers(headers: &HeaderMap, container: &'static FrozenDiContainer) -> Self {
        let trace = extract_trace_context(headers);
        let auth = extract_auth(headers, container).await;
        let tenant = container
            .try_get::<TenantRegistry>()
            .and_then(|tr| tr.resolve(headers));
        Self {
            trace,
            tenant,
            claims: auth.claims,
            session: auth.session,
        }
    }

    /// Consumer-mesh messages: the producing request's identity rides the
    /// envelope. The tenant id is validated against the SAME registry as
    /// HTTP — a suspended tenant's queued events resolve to `None` and stop
    /// being processed — and the producer's `traceparent` continues the
    /// distributed trace instead of starting an orphan root.
    ///
    /// Messages authenticate at the transport (broker credentials), not per
    /// event, so `claims`/`session` are absent by design.
    pub fn from_message(msg: &InboundMessage, container: &'static FrozenDiContainer) -> Self {
        let trace = msg
            .traceparent
            .as_deref()
            .and_then(TraceContext::from_traceparent)
            .unwrap_or_else(TraceContext::new_root);
        let tenant = msg.tenant.as_deref().and_then(|id| {
            container
                .try_get::<TenantRegistry>()
                .and_then(|tr| tr.resolve_by_id(id))
        });
        Self {
            trace,
            tenant,
            claims: None,
            session: None,
        }
    }

    /// `traceparent` for stamping outbound hops (HTTP calls, outbox rows,
    /// queue envelopes).
    pub fn traceparent(&self) -> String {
        self.trace.to_traceparent()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::engine::DiContainerBuilder;
    use crate::web::tenant::{TenantId, TenantStrategy};

    fn container_with_registry() -> &'static FrozenDiContainer {
        let registry = TenantRegistry::new(
            TenantStrategy::header("x-tenant-id"),
            vec![TenantConfig {
                id: TenantId::new("acme"),
                display_name: "Acme".into(),
                datasource: "acme".into(),
            }],
            None,
        );
        registry.suspend(&TenantId::new("globex")); // suspended, never known
        let mut b = DiContainerBuilder::new();
        b.register(registry);
        b.freeze()
    }

    fn msg(tenant: Option<&str>, traceparent: Option<&str>) -> InboundMessage {
        InboundMessage {
            topic: "t".into(),
            payload: serde_json::Value::Null,
            idempotency_key: "k".into(),
            tenant: tenant.map(str::to_owned),
            traceparent: traceparent.map(str::to_owned),
        }
    }

    #[test]
    fn message_provenance_validates_tenant_against_registry() {
        let c = container_with_registry();

        // Known tenant resolves to the same config HTTP would see.
        let p = Provenance::from_message(&msg(Some("acme"), None), c);
        assert_eq!(p.tenant.as_deref().map(|t| t.id.as_str()), Some("acme"));

        // Unknown and suspended ids resolve to None — handlers never see them.
        assert!(Provenance::from_message(&msg(Some("nope"), None), c)
            .tenant
            .is_none());
        assert!(Provenance::from_message(&msg(Some("globex"), None), c)
            .tenant
            .is_none());
        // No tenant on the envelope: simply absent.
        assert!(Provenance::from_message(&msg(None, None), c)
            .tenant
            .is_none());
    }

    #[test]
    fn message_provenance_continues_producer_trace() {
        let c = container_with_registry();
        let carried = "00-0123456789abcdef0123456789abcdef-00f067aa0ba902b7-01";

        let p = Provenance::from_message(&msg(None, Some(carried)), c);
        let hex = crate::observability::lean_telemetry::hex_encode(&p.trace.trace_id);
        assert_eq!(
            hex, "0123456789abcdef0123456789abcdef",
            "trace id continues"
        );
        // Producer's span is this hop's parent; a fresh consumer span is minted.
        let parent = crate::observability::lean_telemetry::hex_encode(&p.trace.parent_span_id);
        assert_eq!(parent, "00f067aa0ba902b7");

        // No traceparent → fresh root (all-zero parent).
        let root = Provenance::from_message(&msg(None, None), c);
        assert_eq!(root.trace.parent_span_id, [0u8; 8]);
    }
}