lenso_service/
direct_http.rs

1use crate::{
2    AuthenticatedServiceContext, AuthenticatedServicePrincipal, AuthenticatedTransportBinding,
3    CallPolicyDeclaration, CallPolicyEvent, CallPolicyEvidence, CallPolicyFailure,
4    CallPolicyRuntime, CallPolicyTerminalOutcome, DelegatedActorContext, DelegatedContextVerifier,
5    EndpointResolver, IdentityDecisionRecorder, ServiceContext, ServiceContextAdmission,
6    ServiceContextPolicy, ServiceReference, TenantContext, WorkloadIdentityProvider,
7    WorkloadIdentityVerification,
8};
9use axum::{
10    Router,
11    body::{Body, Bytes, to_bytes},
12    extract::{Request, State},
13    http::{HeaderMap, HeaderValue, Method, StatusCode, header},
14    response::{IntoResponse, Response},
15    routing::any,
16};
17use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
18use serde::{Deserialize, Serialize};
19use serde_json::{Value, json};
20use std::{
21    collections::BTreeMap,
22    future::Future,
23    pin::Pin,
24    sync::Arc,
25    time::{Duration, SystemTime, UNIX_EPOCH},
26};
27
28const DEADLINE_HEADER: &str = "x-lenso-deadline-unix-ms";
29const IDEMPOTENCY_HEADER: &str = "idempotency-key";
30const AUTHORIZATION_HEADER: &str = "authorization";
31const DELEGATED_ACTOR_HEADER: &str = "x-lenso-delegated-actor";
32const TENANT_CONTEXT_HEADER: &str = "x-lenso-tenant-context";
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(rename_all = "snake_case")]
36pub enum HttpIdempotency {
37    Unknown,
38    Idempotent,
39    RequiresKey,
40}
41
42#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(rename_all = "camelCase")]
44pub struct DirectHttpOperation {
45    pub operation_id: String,
46    pub method: String,
47    pub path: String,
48    pub idempotency: HttpIdempotency,
49    pub call_policy: CallPolicyDeclaration,
50    #[serde(skip_serializing_if = "Option::is_none")]
51    pub request_schema: Option<Value>,
52    pub response_schemas: BTreeMap<String, Value>,
53    #[serde(skip_serializing_if = "Option::is_none")]
54    pub standard_error_schema: Option<Value>,
55}
56
57impl DirectHttpOperation {
58    #[must_use]
59    pub fn no_retry_reason(&self) -> Option<&'static str> {
60        match self.idempotency {
61            HttpIdempotency::Unknown => Some("operation_retry_safety_unknown"),
62            HttpIdempotency::Idempotent | HttpIdempotency::RequiresKey => None,
63        }
64    }
65
66    #[must_use]
67    pub fn retry_decision(&self, status: StatusCode, attempt: u32) -> RetryDecision {
68        self.retry_decision_for(status, attempt, None)
69    }
70
71    fn retry_decision_for(
72        &self,
73        status: StatusCode,
74        attempt: u32,
75        idempotency_key: Option<&str>,
76    ) -> RetryDecision {
77        if self.idempotency == HttpIdempotency::Unknown {
78            return RetryDecision::no("operation_retry_safety_unknown");
79        }
80        if attempt >= self.call_policy.max_attempts {
81            return RetryDecision::no("initial_policy_attempt_limit");
82        }
83        if !matches!(status.as_u16(), 429 | 502 | 503 | 504) {
84            return RetryDecision::no("failure_not_retryable");
85        }
86        match self.idempotency {
87            HttpIdempotency::Idempotent => RetryDecision::yes(),
88            HttpIdempotency::RequiresKey if idempotency_key.is_some_and(|key| !key.is_empty()) => {
89                RetryDecision::yes()
90            }
91            HttpIdempotency::RequiresKey => RetryDecision::no("idempotency_key_required"),
92            HttpIdempotency::Unknown => unreachable!("unknown safety returns before matching"),
93        }
94    }
95}
96
97#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98#[serde(rename_all = "camelCase")]
99pub struct DirectHttpBindings {
100    pub contract_id: String,
101    pub version: String,
102    pub operations: Vec<DirectHttpOperation>,
103}
104
105impl DirectHttpBindings {
106    #[must_use]
107    pub fn operation(&self, operation_id: &str) -> Option<&DirectHttpOperation> {
108        self.operations
109            .iter()
110            .find(|item| item.operation_id == operation_id)
111    }
112
113    fn match_request(&self, method: &Method, path: &str) -> Option<&DirectHttpOperation> {
114        self.operations.iter().find(|operation| {
115            operation.method == method.as_str() && path_matches(&operation.path, path)
116        })
117    }
118}
119
120#[derive(Debug, Clone, PartialEq, Eq)]
121pub struct BindingGenerationError(pub String);
122
123impl std::fmt::Display for BindingGenerationError {
124    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125        formatter.write_str(&self.0)
126    }
127}
128
129impl std::error::Error for BindingGenerationError {}
130
131pub fn generate_direct_http_bindings(
132    contract_id: impl Into<String>,
133    version: impl Into<String>,
134    openapi: &Value,
135) -> Result<DirectHttpBindings, BindingGenerationError> {
136    let version = version.into();
137    let document_version = openapi.pointer("/info/version").and_then(Value::as_str);
138    if document_version != Some(version.as_str()) {
139        return Err(BindingGenerationError(
140            "OpenAPI info.version must match the Service Contract version".to_owned(),
141        ));
142    }
143    let paths = openapi
144        .get("paths")
145        .and_then(Value::as_object)
146        .ok_or_else(|| {
147            BindingGenerationError("OpenAPI Service Contract requires paths".to_owned())
148        })?;
149    let mut operations = Vec::new();
150    for (path, item) in paths {
151        let Some(item) = item.as_object() else {
152            continue;
153        };
154        for method in ["get", "post", "put", "patch", "delete"] {
155            let Some(operation) = item.get(method).and_then(Value::as_object) else {
156                continue;
157            };
158            let operation_id = operation
159                .get("operationId")
160                .and_then(Value::as_str)
161                .ok_or_else(|| {
162                    BindingGenerationError(format!("{method} {path} requires operationId"))
163                })?;
164            let idempotency = match operation.get("x-lenso-idempotency").and_then(Value::as_str) {
165                Some("idempotent") => HttpIdempotency::Idempotent,
166                Some("requires_key") => HttpIdempotency::RequiresKey,
167                Some(value) => {
168                    return Err(BindingGenerationError(format!(
169                        "unsupported x-lenso-idempotency `{value}`"
170                    )));
171                }
172                None => HttpIdempotency::Unknown,
173            };
174            let retry_safe = idempotency != HttpIdempotency::Unknown;
175            let call_policy = operation
176                .get("x-lenso-call-policy")
177                .ok_or_else(|| {
178                    BindingGenerationError(format!("{method} {path} requires x-lenso-call-policy"))
179                })
180                .and_then(|value| {
181                    serde_json::from_value::<CallPolicyDeclaration>(value.clone()).map_err(
182                        |error| {
183                            BindingGenerationError(format!(
184                                "{method} {path} has invalid x-lenso-call-policy: {error}"
185                            ))
186                        },
187                    )
188                })?;
189            if let Some(issue) = call_policy.validate(retry_safe).into_iter().next() {
190                return Err(BindingGenerationError(format!(
191                    "{method} {path} x-lenso-call-policy.{}: {}",
192                    issue.path, issue.code
193                )));
194            }
195            operations.push(DirectHttpOperation {
196                operation_id: operation_id.to_owned(),
197                method: method.to_uppercase(),
198                path: path.clone(),
199                idempotency,
200                call_policy,
201                request_schema: operation
202                    .get("requestBody")
203                    .and_then(|value| value.pointer("/content/application~1json/schema"))
204                    .map(|schema| resolve_local_schema(openapi, schema)),
205                response_schemas: response_schemas(openapi, operation),
206                standard_error_schema: operation
207                    .get("responses")
208                    .and_then(Value::as_object)
209                    .and_then(|responses| {
210                        responses.values().find_map(|response| {
211                            response
212                                .pointer("/content/application~1problem+json/schema")
213                                .map(|schema| resolve_local_schema(openapi, schema))
214                        })
215                    }),
216            });
217        }
218    }
219    operations.sort_by(|left, right| left.operation_id.cmp(&right.operation_id));
220    Ok(DirectHttpBindings {
221        contract_id: contract_id.into(),
222        version,
223        operations,
224    })
225}
226
227#[derive(Debug, Clone)]
228pub struct DirectHttpRequest {
229    pub method: Method,
230    pub path: String,
231    pub headers: HeaderMap,
232    pub body: Bytes,
233    pub deadline_unix_ms: Option<u64>,
234    pub idempotency_key: Option<String>,
235    pub workload_credential: Option<String>,
236    pub authenticated_transport_binding: Option<String>,
237    pub authenticated_service_principal: Option<AuthenticatedServicePrincipal>,
238    pub delegated_actor_context: Option<DelegatedActorContext>,
239    pub tenant_context: Option<TenantContext>,
240    service_context_decode_failed: bool,
241    pub authenticated_service_context: Option<AuthenticatedServiceContext>,
242}
243
244impl DirectHttpRequest {
245    #[must_use]
246    pub fn new(method: Method, path: impl Into<String>) -> Self {
247        Self {
248            method,
249            path: path.into(),
250            headers: HeaderMap::new(),
251            body: Bytes::new(),
252            deadline_unix_ms: None,
253            idempotency_key: None,
254            workload_credential: None,
255            authenticated_transport_binding: None,
256            authenticated_service_principal: None,
257            delegated_actor_context: None,
258            tenant_context: None,
259            service_context_decode_failed: false,
260            authenticated_service_context: None,
261        }
262    }
263
264    #[must_use]
265    pub fn with_deadline(mut self, deadline_unix_ms: u64) -> Self {
266        self.deadline_unix_ms = Some(deadline_unix_ms);
267        self
268    }
269
270    #[must_use]
271    pub fn with_workload_credential(mut self, credential: impl Into<String>) -> Self {
272        self.workload_credential = Some(credential.into());
273        self
274    }
275
276    #[must_use]
277    pub fn with_authenticated_transport_binding(mut self, binding: impl Into<String>) -> Self {
278        self.authenticated_transport_binding = Some(binding.into());
279        self
280    }
281
282    #[must_use]
283    pub fn with_service_context(
284        mut self,
285        actor: DelegatedActorContext,
286        tenant: Option<TenantContext>,
287    ) -> Self {
288        self.delegated_actor_context = Some(actor);
289        self.tenant_context = tenant;
290        self
291    }
292}
293
294#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
295#[serde(rename_all = "camelCase")]
296pub struct DirectHttpEvidence {
297    pub operation_id: Option<String>,
298    pub decision: String,
299    pub call_policy: CallPolicyEvidence,
300    pub native_status: Option<u16>,
301}
302
303#[derive(Debug, Clone)]
304pub struct DirectHttpResponse {
305    pub status: StatusCode,
306    pub headers: HeaderMap,
307    pub body: Bytes,
308    pub standard_error: Option<Value>,
309    pub evidence: Option<DirectHttpEvidence>,
310}
311
312impl DirectHttpResponse {
313    #[must_use]
314    pub fn json(status: StatusCode, body: Value) -> Self {
315        let mut headers = HeaderMap::new();
316        headers.insert(
317            header::CONTENT_TYPE,
318            HeaderValue::from_static("application/json"),
319        );
320        let standard_error =
321            (status.is_client_error() || status.is_server_error()).then(|| body.clone());
322        Self {
323            status,
324            headers,
325            body: Bytes::from(serde_json::to_vec(&body).expect("JSON value must serialize")),
326            standard_error,
327            evidence: None,
328        }
329    }
330
331    fn problem(status: StatusCode, code: &str, detail: &str, operation_id: Option<String>) -> Self {
332        let mut response = Self::json(
333            status,
334            json!({"type":"about:blank","title":detail,"status":status.as_u16(),"detail":detail,"code":code,"request_id":null,"correlation_id":null,"errors":[]}),
335        );
336        response.headers.insert(
337            header::CONTENT_TYPE,
338            HeaderValue::from_static("application/problem+json"),
339        );
340        response.evidence = Some(DirectHttpEvidence {
341            operation_id,
342            decision: code.to_owned(),
343            call_policy: CallPolicyEvidence {
344                events: vec![if code == "overload_rejected" {
345                    CallPolicyEvent::OverloadRejected
346                } else {
347                    CallPolicyEvent::CallFailed
348                }],
349                attempts: 0,
350                terminal_outcome: CallPolicyTerminalOutcome::Rejected,
351                fallback_handler: None,
352            },
353            native_status: Some(status.as_u16()),
354        });
355        response
356    }
357}
358
359type HandlerFuture = Pin<Box<dyn Future<Output = DirectHttpResponse> + Send>>;
360type Handler = dyn Fn(DirectHttpRequest) -> HandlerFuture + Send + Sync;
361
362#[derive(Clone)]
363pub struct DirectHttpServerBinding {
364    inner: Arc<ServerInner>,
365}
366struct ServerInner {
367    bindings: DirectHttpBindings,
368    handler: Arc<Handler>,
369    policy_runtime: CallPolicyRuntime,
370    workload_identity: Option<DirectHttpWorkloadIdentity>,
371    service_context: Option<DirectHttpServiceContext>,
372}
373
374#[derive(Debug, Clone)]
375struct DirectHttpWorkloadIdentity {
376    provider: Arc<dyn WorkloadIdentityProvider>,
377    audience: String,
378}
379
380#[derive(Debug, Clone)]
381struct DirectHttpServiceContext {
382    admission: ServiceContextAdmission,
383    recorder: Arc<dyn IdentityDecisionRecorder>,
384}
385
386impl std::fmt::Debug for DirectHttpServerBinding {
387    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
388        formatter
389            .debug_struct("DirectHttpServerBinding")
390            .field("bindings", &self.inner.bindings)
391            .finish_non_exhaustive()
392    }
393}
394
395impl DirectHttpServerBinding {
396    pub fn new<F, Fut>(
397        bindings: DirectHttpBindings,
398        provider: Arc<dyn WorkloadIdentityProvider>,
399        audience: impl Into<String>,
400        context_provider: Arc<dyn DelegatedContextVerifier>,
401        context_policies: impl IntoIterator<Item = (String, ServiceContextPolicy)>,
402        evidence_recorder: Arc<dyn IdentityDecisionRecorder>,
403        handler: F,
404    ) -> Self
405    where
406        F: Fn(DirectHttpRequest) -> Fut + Send + Sync + 'static,
407        Fut: Future<Output = DirectHttpResponse> + Send + 'static,
408    {
409        Self::new_with_policy_runtime_unchecked(bindings, CallPolicyRuntime::default(), handler)
410            .with_workload_identity(provider, audience)
411            .with_service_context(context_provider, context_policies, evidence_recorder)
412    }
413
414    #[cfg(debug_assertions)]
415    pub fn new_without_workload_identity<F, Fut>(bindings: DirectHttpBindings, handler: F) -> Self
416    where
417        F: Fn(DirectHttpRequest) -> Fut + Send + Sync + 'static,
418        Fut: Future<Output = DirectHttpResponse> + Send + 'static,
419    {
420        Self::new_with_policy_runtime_unchecked(bindings, CallPolicyRuntime::default(), handler)
421    }
422
423    pub fn new_with_policy_runtime<F, Fut>(
424        bindings: DirectHttpBindings,
425        policy_runtime: CallPolicyRuntime,
426        provider: Arc<dyn WorkloadIdentityProvider>,
427        audience: impl Into<String>,
428        context_provider: Arc<dyn DelegatedContextVerifier>,
429        context_policies: impl IntoIterator<Item = (String, ServiceContextPolicy)>,
430        evidence_recorder: Arc<dyn IdentityDecisionRecorder>,
431        handler: F,
432    ) -> Self
433    where
434        F: Fn(DirectHttpRequest) -> Fut + Send + Sync + 'static,
435        Fut: Future<Output = DirectHttpResponse> + Send + 'static,
436    {
437        Self::new_with_policy_runtime_unchecked(bindings, policy_runtime, handler)
438            .with_workload_identity(provider, audience)
439            .with_service_context(context_provider, context_policies, evidence_recorder)
440    }
441
442    #[cfg(debug_assertions)]
443    pub fn new_with_policy_runtime_without_workload_identity<F, Fut>(
444        bindings: DirectHttpBindings,
445        policy_runtime: CallPolicyRuntime,
446        handler: F,
447    ) -> Self
448    where
449        F: Fn(DirectHttpRequest) -> Fut + Send + Sync + 'static,
450        Fut: Future<Output = DirectHttpResponse> + Send + 'static,
451    {
452        Self::new_with_policy_runtime_unchecked(bindings, policy_runtime, handler)
453    }
454
455    fn new_with_policy_runtime_unchecked<F, Fut>(
456        bindings: DirectHttpBindings,
457        policy_runtime: CallPolicyRuntime,
458        handler: F,
459    ) -> Self
460    where
461        F: Fn(DirectHttpRequest) -> Fut + Send + Sync + 'static,
462        Fut: Future<Output = DirectHttpResponse> + Send + 'static,
463    {
464        Self {
465            inner: Arc::new(ServerInner {
466                bindings,
467                handler: Arc::new(move |request| Box::pin(handler(request))),
468                policy_runtime,
469                workload_identity: None,
470                service_context: None,
471            }),
472        }
473    }
474
475    #[must_use]
476    pub fn with_workload_identity(
477        self,
478        provider: Arc<dyn WorkloadIdentityProvider>,
479        audience: impl Into<String>,
480    ) -> Self {
481        Self {
482            inner: Arc::new(ServerInner {
483                bindings: self.inner.bindings.clone(),
484                handler: Arc::clone(&self.inner.handler),
485                policy_runtime: self.inner.policy_runtime.clone(),
486                workload_identity: Some(DirectHttpWorkloadIdentity {
487                    provider,
488                    audience: audience.into(),
489                }),
490                service_context: self.inner.service_context.clone(),
491            }),
492        }
493    }
494
495    #[must_use]
496    pub fn with_service_context<I, S>(
497        self,
498        provider: Arc<dyn DelegatedContextVerifier>,
499        policies: I,
500        recorder: Arc<dyn IdentityDecisionRecorder>,
501    ) -> Self
502    where
503        I: IntoIterator<Item = (S, ServiceContextPolicy)>,
504        S: Into<String>,
505    {
506        Self {
507            inner: Arc::new(ServerInner {
508                bindings: self.inner.bindings.clone(),
509                handler: Arc::clone(&self.inner.handler),
510                policy_runtime: self.inner.policy_runtime.clone(),
511                workload_identity: self.inner.workload_identity.clone(),
512                service_context: Some(DirectHttpServiceContext {
513                    admission: ServiceContextAdmission::new(provider, policies),
514                    recorder,
515                }),
516            }),
517        }
518    }
519
520    pub async fn handle(&self, request: DirectHttpRequest) -> DirectHttpResponse {
521        self.inner.handle(request).await
522    }
523
524    #[must_use]
525    pub fn router(self) -> Router {
526        Router::new()
527            .fallback(any(handle_axum))
528            .with_state(self.inner)
529    }
530}
531
532impl ServerInner {
533    async fn handle(&self, mut request: DirectHttpRequest) -> DirectHttpResponse {
534        let Some(operation) = self.bindings.match_request(&request.method, &request.path) else {
535            return DirectHttpResponse::problem(
536                StatusCode::NOT_FOUND,
537                "operation_not_found",
538                "Operation not found",
539                None,
540            );
541        };
542        if let Some(identity) = &self.workload_identity {
543            let Some(credential) = request.workload_credential.as_deref() else {
544                return DirectHttpResponse::problem(
545                    StatusCode::UNAUTHORIZED,
546                    "workload_identity_required",
547                    "Workload Identity credential is required",
548                    Some(operation.operation_id.clone()),
549                );
550            };
551            let Some(binding) = request.authenticated_transport_binding.as_deref() else {
552                return DirectHttpResponse::problem(
553                    StatusCode::UNAUTHORIZED,
554                    "authenticated_transport_binding_required",
555                    "Authenticated transport binding is required",
556                    Some(operation.operation_id.clone()),
557                );
558            };
559            match identity.provider.verify(
560                credential,
561                &WorkloadIdentityVerification::new(&identity.audience, binding, now_ms()),
562            ) {
563                Ok(principal) => request.authenticated_service_principal = Some(principal),
564                Err(error) => {
565                    return DirectHttpResponse::problem(
566                        StatusCode::UNAUTHORIZED,
567                        &error.evidence.outcome,
568                        &error.message,
569                        Some(operation.operation_id.clone()),
570                    );
571                }
572            }
573        }
574        if let Some(context) = &self.service_context {
575            let decision = if request.service_context_decode_failed {
576                Err(context.admission.invalid_proof(&operation.operation_id))
577            } else {
578                let service_context = request
579                    .delegated_actor_context
580                    .clone()
581                    .map(|actor| ServiceContext::new(actor, request.tenant_context.clone()));
582                context
583                    .admission
584                    .admit(&operation.operation_id, service_context.as_ref(), now_ms())
585            };
586            match decision {
587                Ok(authenticated) => {
588                    if context.recorder.record(&authenticated.evidence).is_err() {
589                        return DirectHttpResponse::problem(
590                            StatusCode::INTERNAL_SERVER_ERROR,
591                            "identity_evidence_persistence_failed",
592                            "Identity decision evidence could not be persisted",
593                            Some(operation.operation_id.clone()),
594                        );
595                    }
596                    request.authenticated_service_context = Some(authenticated);
597                }
598                Err(error) => {
599                    if context.recorder.record(&error.evidence).is_err() {
600                        return DirectHttpResponse::problem(
601                            StatusCode::INTERNAL_SERVER_ERROR,
602                            "identity_evidence_persistence_failed",
603                            "Identity decision evidence could not be persisted",
604                            Some(operation.operation_id.clone()),
605                        );
606                    }
607                    return DirectHttpResponse::problem(
608                        StatusCode::FORBIDDEN,
609                        &error.evidence.outcome,
610                        &error.message,
611                        Some(operation.operation_id.clone()),
612                    );
613                }
614            }
615        }
616        if request
617            .deadline_unix_ms
618            .is_none_or(|deadline| deadline <= now_ms())
619        {
620            return DirectHttpResponse::problem(
621                StatusCode::GATEWAY_TIMEOUT,
622                "deadline_expired",
623                "Deadline is missing or expired",
624                Some(operation.operation_id.clone()),
625            );
626        }
627        if operation.idempotency == HttpIdempotency::RequiresKey
628            && request.idempotency_key.as_deref().is_none_or(str::is_empty)
629        {
630            return DirectHttpResponse::problem(
631                StatusCode::BAD_REQUEST,
632                "idempotency_key_required",
633                "Idempotency Key is required",
634                Some(operation.operation_id.clone()),
635            );
636        }
637        let operation_key = format!("{}:{}", self.bindings.contract_id, operation.operation_id);
638        let Ok(_admission) = self
639            .policy_runtime
640            .admit(operation_key, &operation.call_policy)
641        else {
642            return DirectHttpResponse::problem(
643                StatusCode::TOO_MANY_REQUESTS,
644                "overload_rejected",
645                "Service operation is overloaded",
646                Some(operation.operation_id.clone()),
647            );
648        };
649        (self.handler)(request).await
650    }
651}
652
653async fn handle_axum(State(inner): State<Arc<ServerInner>>, request: Request) -> Response {
654    let (parts, body) = request.into_parts();
655    let deadline_unix_ms = parts
656        .headers
657        .get(DEADLINE_HEADER)
658        .and_then(|value| value.to_str().ok())
659        .and_then(|value| value.parse().ok());
660    let idempotency_key = parts
661        .headers
662        .get(IDEMPOTENCY_HEADER)
663        .and_then(|value| value.to_str().ok())
664        .map(str::to_owned);
665    let workload_credential = parts
666        .headers
667        .get(AUTHORIZATION_HEADER)
668        .and_then(|value| value.to_str().ok())
669        .and_then(|value| value.strip_prefix("Bearer "))
670        .map(str::to_owned);
671    let authenticated_transport_binding = parts
672        .extensions
673        .get::<AuthenticatedTransportBinding>()
674        .map(|binding| binding.0.clone());
675    let actor =
676        decode_context_header::<DelegatedActorContext>(&parts.headers, DELEGATED_ACTOR_HEADER);
677    let tenant = decode_context_header::<TenantContext>(&parts.headers, TENANT_CONTEXT_HEADER);
678    let service_context_decode_failed = actor.is_err() || tenant.is_err();
679    let delegated_actor_context = actor.ok().flatten();
680    let tenant_context = tenant.ok().flatten();
681    let body = to_bytes(body, 16 * 1024 * 1024).await.unwrap_or_default();
682    let response = inner
683        .handle(DirectHttpRequest {
684            method: parts.method,
685            path: parts.uri.path().to_owned(),
686            headers: parts.headers,
687            body,
688            deadline_unix_ms,
689            idempotency_key,
690            workload_credential,
691            authenticated_transport_binding,
692            authenticated_service_principal: None,
693            delegated_actor_context,
694            tenant_context,
695            service_context_decode_failed,
696            authenticated_service_context: None,
697        })
698        .await;
699    response.into_response()
700}
701
702impl IntoResponse for DirectHttpResponse {
703    fn into_response(self) -> Response {
704        let mut response = Response::new(Body::from(self.body));
705        *response.status_mut() = self.status;
706        *response.headers_mut() = self.headers;
707        response
708    }
709}
710
711#[derive(Debug, Clone)]
712pub struct DirectHttpCall {
713    operation_id: String,
714    path_parameters: BTreeMap<String, String>,
715    body: Option<Value>,
716    deadline_unix_ms: Option<u64>,
717    idempotency_key: Option<String>,
718    workload_credential: Option<String>,
719    delegated_actor_context: Option<DelegatedActorContext>,
720    tenant_context: Option<TenantContext>,
721}
722impl DirectHttpCall {
723    #[must_use]
724    pub fn new(operation_id: impl Into<String>) -> Self {
725        Self {
726            operation_id: operation_id.into(),
727            path_parameters: BTreeMap::new(),
728            body: None,
729            deadline_unix_ms: None,
730            idempotency_key: None,
731            workload_credential: None,
732            delegated_actor_context: None,
733            tenant_context: None,
734        }
735    }
736    #[must_use]
737    pub fn with_path_parameter(
738        mut self,
739        name: impl Into<String>,
740        value: impl Into<String>,
741    ) -> Self {
742        self.path_parameters.insert(name.into(), value.into());
743        self
744    }
745    #[must_use]
746    pub fn with_json(mut self, body: Value) -> Self {
747        self.body = Some(body);
748        self
749    }
750    #[must_use]
751    pub fn with_deadline(mut self, deadline: u64) -> Self {
752        self.deadline_unix_ms = Some(deadline);
753        self
754    }
755    #[must_use]
756    pub fn with_idempotency_key(mut self, key: impl Into<String>) -> Self {
757        self.idempotency_key = Some(key.into());
758        self
759    }
760
761    #[must_use]
762    pub fn with_workload_credential(mut self, credential: impl Into<String>) -> Self {
763        self.workload_credential = Some(credential.into());
764        self
765    }
766
767    #[must_use]
768    pub fn with_service_context(
769        mut self,
770        actor: DelegatedActorContext,
771        tenant: Option<TenantContext>,
772    ) -> Self {
773        self.delegated_actor_context = Some(actor);
774        self.tenant_context = tenant;
775        self
776    }
777}
778
779pub struct DirectHttpClient<R> {
780    resolver: R,
781    bindings: DirectHttpBindings,
782    http: reqwest::Client,
783    policy_runtime: CallPolicyRuntime,
784    fallbacks: BTreeMap<String, Arc<HttpFallback>>,
785}
786type HttpFallback = dyn Fn(CallPolicyFailure) -> DirectHttpResponse + Send + Sync;
787impl<R> std::fmt::Debug for DirectHttpClient<R> {
788    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
789        formatter
790            .debug_struct("DirectHttpClient")
791            .field("bindings", &self.bindings)
792            .field("policy_runtime", &self.policy_runtime)
793            .field("fallbacks", &self.fallbacks.keys().collect::<Vec<_>>())
794            .finish_non_exhaustive()
795    }
796}
797impl<R: EndpointResolver> DirectHttpClient<R> {
798    #[must_use]
799    pub fn new(resolver: R, bindings: DirectHttpBindings) -> Self {
800        Self {
801            resolver,
802            bindings,
803            http: reqwest::Client::new(),
804            policy_runtime: CallPolicyRuntime::default(),
805            fallbacks: BTreeMap::new(),
806        }
807    }
808
809    #[must_use]
810    pub fn with_policy_runtime(mut self, policy_runtime: CallPolicyRuntime) -> Self {
811        self.policy_runtime = policy_runtime;
812        self
813    }
814
815    #[must_use]
816    pub fn with_fallback<F>(mut self, handler: impl Into<String>, fallback: F) -> Self
817    where
818        F: Fn(CallPolicyFailure) -> DirectHttpResponse + Send + Sync + 'static,
819    {
820        self.fallbacks.insert(handler.into(), Arc::new(fallback));
821        self
822    }
823    pub async fn call(
824        &self,
825        service: &ServiceReference,
826        call: DirectHttpCall,
827    ) -> Result<DirectHttpResponse, DirectHttpCallError> {
828        let operation = self.bindings.operation(&call.operation_id).ok_or_else(|| {
829            DirectHttpCallError::Contract(
830                "operation is not declared by the generated binding".to_owned(),
831            )
832        })?;
833        let deadline = call
834            .deadline_unix_ms
835            .ok_or_else(|| DirectHttpCallError::Contract("deadline_required".to_owned()))?;
836        if deadline <= now_ms() {
837            return self.deadline_failure(operation, 0, Vec::new());
838        }
839        if operation.idempotency == HttpIdempotency::RequiresKey
840            && call.idempotency_key.as_deref().is_none_or(str::is_empty)
841        {
842            return Err(DirectHttpCallError::Contract(
843                "idempotency_key_required".to_owned(),
844            ));
845        }
846        let method = Method::from_bytes(operation.method.as_bytes())
847            .map_err(|error| DirectHttpCallError::Contract(error.to_string()))?;
848        let path = expand_path(&operation.path, &call.path_parameters)?;
849        let operation_key = format!(
850            "{}:{}:{}",
851            service.as_str(),
852            self.bindings.contract_id,
853            operation.operation_id
854        );
855        let permit = match self
856            .policy_runtime
857            .begin_call(operation_key, &operation.call_policy)
858        {
859            Ok(permit) => permit,
860            Err(event) => {
861                let failure = match event {
862                    CallPolicyEvent::CircuitOpen => CallPolicyFailure::CircuitOpen,
863                    CallPolicyEvent::BulkheadSaturated => CallPolicyFailure::BulkheadSaturated,
864                    _ => CallPolicyFailure::NonRetryableFailure,
865                };
866                if let Some(response) =
867                    self.fallback_response(operation, failure, 0, vec![event], None)
868                {
869                    return Ok(response);
870                }
871                return Err(DirectHttpCallError::Policy {
872                    failure,
873                    evidence: CallPolicyEvidence {
874                        events: vec![event],
875                        attempts: 0,
876                        terminal_outcome: CallPolicyTerminalOutcome::Rejected,
877                        fallback_handler: None,
878                    },
879                });
880            }
881        };
882        let state = self
883            .resolver
884            .resolve(service)
885            .map_err(|error| DirectHttpCallError::Resolution(error.to_string()))?;
886        let endpoint = state
887            .endpoints
888            .first()
889            .ok_or_else(|| DirectHttpCallError::Resolution("no usable endpoint".to_owned()))?;
890        let url = format!("{}{}", endpoint.address.trim_end_matches('/'), path);
891        let mut retry_events = Vec::new();
892        for attempt in 1..=operation.call_policy.max_attempts {
893            let remaining_ms = deadline.saturating_sub(now_ms());
894            if remaining_ms == 0 {
895                retry_events.push(CallPolicyEvent::DeadlineExpired);
896                let events = permit.failure_after(retry_events);
897                return self.deadline_failure(operation, attempt - 1, events);
898            }
899            let mut request = self
900                .http
901                .request(method.clone(), &url)
902                .timeout(Duration::from_millis(remaining_ms))
903                .header(DEADLINE_HEADER, deadline);
904            if let Some(key) = call.idempotency_key.as_deref() {
905                request = request.header(IDEMPOTENCY_HEADER, key);
906            }
907            if let Some(credential) = call.workload_credential.as_deref() {
908                request = request.bearer_auth(credential);
909            }
910            if let Some(actor) = call.delegated_actor_context.as_ref() {
911                request = request.header(DELEGATED_ACTOR_HEADER, encode_context_header(actor)?);
912            }
913            if let Some(tenant) = call.tenant_context.as_ref() {
914                request = request.header(TENANT_CONTEXT_HEADER, encode_context_header(tenant)?);
915            }
916            if let Some(body) = call.body.as_ref() {
917                request = request.json(body);
918            }
919            let response = match request.send().await {
920                Ok(response) => response,
921                Err(error) => {
922                    let failure = if error.is_timeout() {
923                        retry_events.push(CallPolicyEvent::DeadlineExpired);
924                        CallPolicyFailure::DeadlineExpired
925                    } else {
926                        retry_events.push(CallPolicyEvent::CallFailed);
927                        CallPolicyFailure::TransportFailure
928                    };
929                    let events = permit.failure_after(retry_events);
930                    if let Some(response) =
931                        self.fallback_response(operation, failure, attempt, events.clone(), None)
932                    {
933                        return Ok(response);
934                    }
935                    return Err(DirectHttpCallError::Transport {
936                        message: format!("transport_failure_no_retry: {error}"),
937                        evidence: CallPolicyEvidence {
938                            events,
939                            attempts: attempt,
940                            terminal_outcome: CallPolicyTerminalOutcome::Failed,
941                            fallback_handler: None,
942                        },
943                    });
944                }
945            };
946            let status = response.status();
947            let decision =
948                operation.retry_decision_for(status, attempt, call.idempotency_key.as_deref());
949            if decision.should_retry {
950                retry_events.push(CallPolicyEvent::RetryScheduled);
951                continue;
952            }
953            let headers = response.headers().clone();
954            let body = match response.bytes().await {
955                Ok(body) => body,
956                Err(error) => {
957                    retry_events.push(CallPolicyEvent::CallFailed);
958                    let events = permit.failure_after(retry_events);
959                    return Err(DirectHttpCallError::Transport {
960                        message: error.to_string(),
961                        evidence: CallPolicyEvidence {
962                            events,
963                            attempts: attempt,
964                            terminal_outcome: CallPolicyTerminalOutcome::Failed,
965                            fallback_handler: None,
966                        },
967                    });
968                }
969            };
970            let standard_error = serde_json::from_slice(&body).ok().filter(|value| {
971                is_standard_problem(value, status, operation.standard_error_schema.as_ref())
972            });
973            let retryable_failure = matches!(status.as_u16(), 429 | 502 | 503 | 504);
974            if status == StatusCode::TOO_MANY_REQUESTS {
975                retry_events.push(CallPolicyEvent::OverloadRejected);
976            }
977            retry_events.push(if status.is_success() {
978                CallPolicyEvent::CallCompleted
979            } else {
980                CallPolicyEvent::CallFailed
981            });
982            let events = if retryable_failure {
983                permit.failure_after(retry_events)
984            } else {
985                permit.success_after(retry_events)
986            };
987            let failure = if status == StatusCode::TOO_MANY_REQUESTS {
988                CallPolicyFailure::OverloadRejected
989            } else if retryable_failure {
990                CallPolicyFailure::RetryableFailure
991            } else {
992                CallPolicyFailure::NonRetryableFailure
993            };
994            if !status.is_success() {
995                if let Some(response) = self.fallback_response(
996                    operation,
997                    failure,
998                    attempt,
999                    events.clone(),
1000                    Some(status),
1001                ) {
1002                    return Ok(response);
1003                }
1004            }
1005            let evidence = Some(DirectHttpEvidence {
1006                operation_id: Some(operation.operation_id.clone()),
1007                decision: if status.is_success() {
1008                    "call_completed".to_owned()
1009                } else {
1010                    decision.reason.to_owned()
1011                },
1012                call_policy: CallPolicyEvidence {
1013                    events,
1014                    attempts: attempt,
1015                    terminal_outcome: if status.is_success() {
1016                        CallPolicyTerminalOutcome::Completed
1017                    } else {
1018                        CallPolicyTerminalOutcome::Failed
1019                    },
1020                    fallback_handler: None,
1021                },
1022                native_status: Some(status.as_u16()),
1023            });
1024            return Ok(DirectHttpResponse {
1025                status,
1026                headers,
1027                body,
1028                standard_error,
1029                evidence,
1030            });
1031        }
1032        unreachable!("the direct HTTP call loop always returns by its final attempt")
1033    }
1034
1035    fn fallback_response(
1036        &self,
1037        operation: &DirectHttpOperation,
1038        failure: CallPolicyFailure,
1039        attempts: u32,
1040        mut events: Vec<CallPolicyEvent>,
1041        native_status: Option<StatusCode>,
1042    ) -> Option<DirectHttpResponse> {
1043        let declaration = operation.call_policy.fallback_for(failure)?;
1044        let fallback = self.fallbacks.get(&declaration.handler)?;
1045        let mut response = fallback(failure);
1046        events.push(CallPolicyEvent::FallbackApplied);
1047        response.evidence = Some(DirectHttpEvidence {
1048            operation_id: Some(operation.operation_id.clone()),
1049            decision: "fallback_applied".to_owned(),
1050            call_policy: CallPolicyEvidence {
1051                events,
1052                attempts,
1053                terminal_outcome: CallPolicyTerminalOutcome::Fallback,
1054                fallback_handler: Some(declaration.handler.clone()),
1055            },
1056            native_status: native_status.map(|status| status.as_u16()),
1057        });
1058        Some(response)
1059    }
1060
1061    fn deadline_failure(
1062        &self,
1063        operation: &DirectHttpOperation,
1064        attempts: u32,
1065        mut events: Vec<CallPolicyEvent>,
1066    ) -> Result<DirectHttpResponse, DirectHttpCallError> {
1067        if !events.contains(&CallPolicyEvent::DeadlineExpired) {
1068            events.push(CallPolicyEvent::DeadlineExpired);
1069        }
1070        if let Some(response) = self.fallback_response(
1071            operation,
1072            CallPolicyFailure::DeadlineExpired,
1073            attempts,
1074            events.clone(),
1075            None,
1076        ) {
1077            return Ok(response);
1078        }
1079        Err(DirectHttpCallError::Policy {
1080            failure: CallPolicyFailure::DeadlineExpired,
1081            evidence: CallPolicyEvidence {
1082                events,
1083                attempts,
1084                terminal_outcome: CallPolicyTerminalOutcome::Failed,
1085                fallback_handler: None,
1086            },
1087        })
1088    }
1089}
1090
1091fn encode_context_header<T: Serialize>(value: &T) -> Result<String, DirectHttpCallError> {
1092    serde_json::to_vec(value)
1093        .map(|json| URL_SAFE_NO_PAD.encode(json))
1094        .map_err(|error| DirectHttpCallError::Contract(error.to_string()))
1095}
1096
1097fn decode_context_header<T: for<'de> Deserialize<'de>>(
1098    headers: &HeaderMap,
1099    name: &str,
1100) -> Result<Option<T>, ()> {
1101    let Some(value) = headers.get(name) else {
1102        return Ok(None);
1103    };
1104    value
1105        .to_str()
1106        .map_err(|_| ())
1107        .and_then(|value| URL_SAFE_NO_PAD.decode(value).map_err(|_| ()))
1108        .and_then(|value| serde_json::from_slice(&value).map_err(|_| ()))
1109        .map(Some)
1110}
1111
1112#[derive(Debug, Clone, PartialEq, Eq)]
1113pub enum DirectHttpCallError {
1114    Contract(String),
1115    Resolution(String),
1116    Transport {
1117        message: String,
1118        evidence: CallPolicyEvidence,
1119    },
1120    Policy {
1121        failure: CallPolicyFailure,
1122        evidence: CallPolicyEvidence,
1123    },
1124}
1125impl std::fmt::Display for DirectHttpCallError {
1126    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1127        match self {
1128            Self::Contract(value) | Self::Resolution(value) => f.write_str(value),
1129            Self::Transport { message, .. } => f.write_str(message),
1130            Self::Policy { failure, .. } => write!(f, "call policy rejected: {failure:?}"),
1131        }
1132    }
1133}
1134impl std::error::Error for DirectHttpCallError {}
1135
1136#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1137pub struct RetryDecision {
1138    pub should_retry: bool,
1139    pub reason: &'static str,
1140}
1141impl RetryDecision {
1142    pub(crate) fn yes() -> Self {
1143        Self {
1144            should_retry: true,
1145            reason: "declared_safe_retry",
1146        }
1147    }
1148    pub(crate) fn no(reason: &'static str) -> Self {
1149        Self {
1150            should_retry: false,
1151            reason,
1152        }
1153    }
1154}
1155
1156fn path_matches(template: &str, actual: &str) -> bool {
1157    let template = template.trim_matches('/').split('/');
1158    let actual = actual.trim_matches('/').split('/');
1159    let template: Vec<_> = template.collect();
1160    let actual: Vec<_> = actual.collect();
1161    template.len() == actual.len()
1162        && template.iter().zip(actual).all(|(expected, value)| {
1163            (expected.starts_with('{') && expected.ends_with('}')) || *expected == value
1164        })
1165}
1166
1167fn expand_path(
1168    template: &str,
1169    parameters: &BTreeMap<String, String>,
1170) -> Result<String, DirectHttpCallError> {
1171    let mut path = String::new();
1172    for segment in template.trim_start_matches('/').split('/') {
1173        path.push('/');
1174        if let Some(name) = segment
1175            .strip_prefix('{')
1176            .and_then(|value| value.strip_suffix('}'))
1177        {
1178            let value = parameters
1179                .get(name)
1180                .filter(|value| !value.is_empty())
1181                .ok_or_else(|| {
1182                    DirectHttpCallError::Contract(format!("missing path parameter `{name}`"))
1183                })?;
1184            if value.contains('/') {
1185                return Err(DirectHttpCallError::Contract(format!(
1186                    "path parameter `{name}` must be one segment"
1187                )));
1188            }
1189            path.push_str(&encode_path_segment(value));
1190        } else if !segment.is_empty() {
1191            path.push_str(segment);
1192        }
1193    }
1194    Ok(path)
1195}
1196
1197fn encode_path_segment(value: &str) -> String {
1198    let mut encoded = String::new();
1199    for byte in value.as_bytes() {
1200        if byte.is_ascii_alphanumeric() || matches!(*byte, b'-' | b'.' | b'_' | b'~') {
1201            encoded.push(char::from(*byte));
1202        } else {
1203            encoded.push_str(&format!("%{byte:02X}"));
1204        }
1205    }
1206    encoded
1207}
1208
1209fn is_standard_problem(value: &Value, status: StatusCode, schema: Option<&Value>) -> bool {
1210    let Some(required) = schema
1211        .and_then(|schema| schema.get("required"))
1212        .and_then(Value::as_array)
1213    else {
1214        return false;
1215    };
1216    required
1217        .iter()
1218        .filter_map(Value::as_str)
1219        .all(|field| value.get(field).is_some())
1220        && value.get("status").and_then(Value::as_u64) == Some(u64::from(status.as_u16()))
1221}
1222
1223fn resolve_local_schema(openapi: &Value, schema: &Value) -> Value {
1224    schema
1225        .get("$ref")
1226        .and_then(Value::as_str)
1227        .and_then(|reference| reference.strip_prefix('#'))
1228        .and_then(|pointer| openapi.pointer(pointer))
1229        .cloned()
1230        .unwrap_or_else(|| schema.clone())
1231}
1232
1233fn response_schemas(
1234    openapi: &Value,
1235    operation: &serde_json::Map<String, Value>,
1236) -> BTreeMap<String, Value> {
1237    operation
1238        .get("responses")
1239        .and_then(Value::as_object)
1240        .into_iter()
1241        .flatten()
1242        .filter_map(|(status, response)| {
1243            response
1244                .get("content")
1245                .and_then(Value::as_object)
1246                .and_then(|content| content.values().find_map(|media| media.get("schema")))
1247                .map(|schema| (status.clone(), resolve_local_schema(openapi, schema)))
1248        })
1249        .collect()
1250}
1251
1252fn now_ms() -> u64 {
1253    SystemTime::now()
1254        .duration_since(UNIX_EPOCH)
1255        .unwrap_or_default()
1256        .as_millis() as u64
1257}
lenso_service/direct_http.rs

lenso_service/
direct_http.rs
