platform_http/
context.rs

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

platform_http/
context.rs
