platform_http/
context.rs

1use axum::body::Body;
2use axum::extract::State;
3use axum::extract::{FromRequestParts, Request};
4use axum::http::header::HeaderName;
5use axum::http::request::Parts;
6use axum::middleware::Next;
7use axum::response::Response;
8use chrono::{DateTime, Utc};
9use platform_core::{
10    ActorResolutionRequest, AppContext, CorrelationId, IdGenerator, RequestContext, RequestId,
11    UuidGenerator, generate_trace_context,
12    story_events::{
13        HttpRequestStoryEventRecord, http_request_story_creation, http_request_story_event_id,
14        insert_http_request_story_projection,
15    },
16    trace_context_from_traceparent,
17};
18use std::time::Instant;
19use tracing::Instrument;
20
21const REQUEST_ID_HEADER: &str = "x-request-id";
22const CORRELATION_ID_HEADER: &str = "x-correlation-id";
23const AUTHORIZATION_HEADER: &str = "authorization";
24const COOKIE_HEADER: &str = "cookie";
25const TRACEPARENT_HEADER: &str = "traceparent";
26
27#[derive(Debug, Clone)]
28pub struct HttpRequestContext(pub RequestContext);
29
30impl std::ops::Deref for HttpRequestContext {
31    type Target = RequestContext;
32
33    fn deref(&self) -> &Self::Target {
34        &self.0
35    }
36}
37
38pub async fn request_context_middleware(
39    State(ctx): State<AppContext>,
40    mut request: Request<Body>,
41    next: Next,
42) -> Response {
43    let started_at = Instant::now();
44    let started_at_utc = Utc::now();
45    let request_id = header_value(request.headers(), REQUEST_ID_HEADER)
46        .unwrap_or_else(|| UuidGenerator.new_id("req"));
47    let correlation_id = header_value(request.headers(), CORRELATION_ID_HEADER)
48        .unwrap_or_else(|| UuidGenerator.new_id("corr"));
49    let actor = ctx
50        .actor_resolver
51        .resolve_actor(ActorResolutionRequest {
52            authorization: authorization_header(request.headers()),
53            cookie: cookie_header(request.headers()),
54        })
55        .await;
56    let trace = traceparent_header(request.headers())
57        .and_then(|value| trace_context_from_traceparent(&value))
58        .unwrap_or_else(generate_trace_context);
59    let method = request.method().clone();
60    let path = request.uri().path().to_owned();
61
62    let mut context = RequestContext {
63        request_id: RequestId::new(request_id),
64        correlation_id: CorrelationId::new(correlation_id),
65        trace,
66        actor,
67        tenant_id: None,
68        causation_id: None,
69    };
70    context.causation_id = Some(http_request_story_event_id(&context));
71
72    request
73        .extensions_mut()
74        .insert(HttpRequestContext(context.clone()));
75
76    let span = tracing::info_span!(
77        "http_request",
78        request_id = %context.request_id.0,
79        correlation_id = %context.correlation_id.0,
80        lenso.correlation_id = %context.correlation_id.0,
81        lenso.story_id = %context.correlation_id.0,
82        lenso.execution.kind = "http_request",
83        lenso.execution.name = %format!("{} {}", method.as_str(), path.as_str()),
84        otel.trace_id = context.trace.trace_id.as_deref().unwrap_or(""),
85        otel.parent_span_id = context.trace.span_id.as_deref().unwrap_or(""),
86        http_method = %method,
87        http_path = %path,
88    );
89
90    let mut response = next.run(request).instrument(span).await;
91    record_http_request_story(
92        ctx.db.clone(),
93        context.clone(),
94        method.as_str(),
95        path.as_str(),
96        response.status().as_u16(),
97        response.headers().get("x-lenso-error-code"),
98        started_at,
99        started_at_utc,
100    );
101    response.headers_mut().insert(
102        REQUEST_ID_HEADER,
103        context.request_id.0.parse().expect("valid request id"),
104    );
105    response.headers_mut().insert(
106        CORRELATION_ID_HEADER,
107        context
108            .correlation_id
109            .0
110            .parse()
111            .expect("valid correlation id"),
112    );
113    response
114}
115
116fn record_http_request_story(
117    pool: platform_core::DbPool,
118    request_ctx: RequestContext,
119    method: &str,
120    path: &str,
121    status_code: u16,
122    error_code_header: Option<&axum::http::HeaderValue>,
123    started_at: Instant,
124    started_at_utc: DateTime<Utc>,
125) {
126    let error_code = error_code_header
127        .and_then(|value| value.to_str().ok())
128        .filter(|value| !value.is_empty())
129        .map(ToOwned::to_owned);
130    let duration_ms = started_at.elapsed().as_millis().min(i64::MAX as u128) as i64;
131    let record = HttpRequestStoryEventRecord {
132        method: method.to_owned(),
133        path: path.to_owned(),
134        status_code,
135        error_code,
136        creation: http_request_story_creation(path, status_code),
137        started_at: started_at_utc,
138        completed_at: Utc::now(),
139        duration_ms,
140    };
141
142    tokio::spawn(async move {
143        if let Err(error) = insert_http_request_story_projection(&pool, &request_ctx, record).await
144        {
145            tracing::warn!(
146                error = ?error,
147                request_id = %request_ctx.request_id.0,
148                correlation_id = %request_ctx.correlation_id.0,
149                "failed to write HTTP request story projection"
150            );
151        }
152    });
153}
154
155impl<S> FromRequestParts<S> for HttpRequestContext
156where
157    S: Send + Sync,
158{
159    type Rejection = crate::ApiErrorResponse;
160
161    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
162        parts
163            .extensions
164            .get::<HttpRequestContext>()
165            .cloned()
166            .ok_or_else(|| {
167                platform_core::AppError::new(
168                    platform_core::ErrorCode::Internal,
169                    "Request context is missing",
170                )
171                .into()
172            })
173    }
174}
175
176fn header_value(headers: &axum::http::HeaderMap, name: &str) -> Option<String> {
177    let name = HeaderName::from_static(match name {
178        REQUEST_ID_HEADER => REQUEST_ID_HEADER,
179        CORRELATION_ID_HEADER => CORRELATION_ID_HEADER,
180        _ => unreachable!("known request context header"),
181    });
182
183    headers
184        .get(name)
185        .and_then(|value| value.to_str().ok())
186        .filter(|value| !value.is_empty())
187        .map(ToOwned::to_owned)
188}
189
190fn authorization_header(headers: &axum::http::HeaderMap) -> Option<String> {
191    let name = HeaderName::from_static(AUTHORIZATION_HEADER);
192    headers
193        .get(name)
194        .and_then(|value| value.to_str().ok())
195        .filter(|value| !value.is_empty())
196        .map(ToOwned::to_owned)
197}
198
199fn cookie_header(headers: &axum::http::HeaderMap) -> Option<String> {
200    let name = HeaderName::from_static(COOKIE_HEADER);
201    headers
202        .get(name)
203        .and_then(|value| value.to_str().ok())
204        .filter(|value| !value.is_empty())
205        .map(ToOwned::to_owned)
206}
207
208fn traceparent_header(headers: &axum::http::HeaderMap) -> Option<String> {
209    let name = HeaderName::from_static(TRACEPARENT_HEADER);
210    headers
211        .get(name)
212        .and_then(|value| value.to_str().ok())
213        .filter(|value| !value.is_empty())
214        .map(ToOwned::to_owned)
215}
platform_http/context.rs

platform_http/
context.rs
