siderite_core/
di.rs

1//! Dependency injection (`Depends`), inspired by FastAPI.
2//!
3//! A [`Dependency`] is any type that knows how to build itself from the
4//! request ([`Dependency::resolve`]). Handlers ask for one with
5//! [`Depends<T>`]; dependencies may in turn depend on others through
6//! [`ResolveContext::resolve`].
7//!
8//! # Scoping and caching
9//! Resolution is **request-scoped and cached**: within one request the same
10//! `T` is resolved at most once, no matter how many handler arguments or other
11//! dependencies ask for it. Failed resolutions are not cached.
12//!
13//! # Application-scoped values, overrides and global dependencies
14//! * [`App::provide`] registers a shared value. It takes precedence over
15//!   `T::resolve` for `Depends<T>` (when `T: Dependency`) and is the only way
16//!   to inject types that are not dependencies, via [`Provided<T>`].
17//! * [`App::override_dependency`] / [`App::override_value`] replace
18//!   `T::resolve` (also when `T` is requested by another dependency); meant
19//!   for tests.
20//! * [`App::dependency`] runs a dependency before **every routed request of
21//!   the app and its mounts** and short-circuits with its error (e.g. 401).
22//!   Unrouted requests (404) and the generated documentation endpoints are not
23//!   affected. Per-route dependencies are not provided; add a `Depends<T>`
24//!   argument to the handler instead.
25//!
26//! A mounted child app sees its parent's registrations; its own registrations
27//! take precedence inside the child.
28//!
29//! # Teardown
30//! `Drop` cannot be async, so cleanup that must await is registered with
31//! [`ResolveContext::on_teardown`] (the analogue of FastAPI's `yield`).
32//! Teardown futures run **after the response has been produced**, in
33//! **reverse (LIFO) registration order**, on a spawned task; errors and panics
34//! are logged. Teardown is best-effort and not tied to the end of response
35//! *streaming*: a streaming body may still be sent while teardown runs.
36//!
37//! # Documentation
38//! [`Dependency::describe`] contributes OpenAPI parameters; `Depends<T>`
39//! forwards to it, so headers/queries of nested dependencies show up in the
40//! operation (cycles are guarded).
41
42use crate::app::App;
43use crate::body::Body;
44use crate::error::ApiError;
45use crate::extract::FromRequestParts;
46use crate::response::IntoResponse;
47use http::request::Parts;
48use siderite_openapi::{Operation, SchemaRegistry};
49use std::any::{Any, TypeId, type_name};
50use std::cell::RefCell;
51use std::collections::HashMap;
52use std::convert::Infallible;
53use std::fmt::Display;
54use std::future::Future;
55use std::ops::Deref;
56use std::pin::Pin;
57use std::sync::{Arc, Mutex, PoisonError};
58use std::task::{Context, Poll};
59use thiserror::Error;
60use tower::{Layer, Service, ServiceExt};
61
62type BoxFut<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
63type Erased = Arc<dyn Any + Send + Sync>;
64
65/// Failure inside the dependency machinery. Rendered to clients as an opaque
66/// 500; the details are logged.
67#[derive(Debug, Clone, Error)]
68#[non_exhaustive]
69pub enum DependencyError {
70    /// A dependency (transitively) depends on itself.
71    #[error("dependency cycle detected: {}", .chain.join(" -> "))]
72    Cycle {
73        /// Type names along the cycle, starting and ending with the same type.
74        chain: Vec<&'static str>,
75    },
76    /// A value was requested with [`Provided`] but never registered.
77    #[error("no value of type `{type_name}` was provided with App::provide")]
78    NotProvided {
79        /// The requested type.
80        type_name: &'static str,
81    },
82}
83
84/// A type that can be built from a request and injected with [`Depends`].
85pub trait Dependency: Send + Sync + Sized + 'static {
86    /// Build the value. Use `ctx` to read the request or resolve other
87    /// dependencies.
88    fn resolve(ctx: &mut ResolveContext<'_>)
89    -> impl Future<Output = Result<Self, ApiError>> + Send;
90
91    /// Document this dependency on `op` (parameters, security, ...).
92    fn describe(_op: &mut Operation, _registry: &mut SchemaRegistry) {}
93}
94
95/// Owned snapshot of the request head handed to dependency overrides.
96#[derive(Debug, Clone)]
97pub struct RequestHead {
98    /// Request method.
99    pub method: http::Method,
100    /// Request URI.
101    pub uri: http::Uri,
102    /// Request headers.
103    pub headers: http::HeaderMap,
104}
105
106impl From<&Parts> for RequestHead {
107    fn from(parts: &Parts) -> Self {
108        Self {
109            method: parts.method.clone(),
110            uri: parts.uri.clone(),
111            headers: parts.headers.clone(),
112        }
113    }
114}
115
116/// What a teardown future may return: `()` or a `Result<(), E>` whose error is
117/// logged.
118pub trait TeardownOutput: Send + 'static {
119    /// The failure message, if the teardown failed.
120    fn failure(self) -> Option<String>;
121}
122
123impl TeardownOutput for () {
124    fn failure(self) -> Option<String> {
125        None
126    }
127}
128
129impl<E: Display + Send + 'static> TeardownOutput for Result<(), E> {
130    fn failure(self) -> Option<String> {
131        self.err().map(|e| e.to_string())
132    }
133}
134
135type Teardown = BoxFut<'static, ()>;
136
137#[derive(Default)]
138struct ScopeState {
139    cache: HashMap<TypeId, Erased>,
140    stack: Vec<(TypeId, &'static str)>,
141    teardowns: Vec<Teardown>,
142}
143
144/// Per-request resolution state, stored in the request extensions.
145#[derive(Clone, Default)]
146pub(crate) struct RequestScope(Arc<Mutex<ScopeState>>);
147
148impl RequestScope {
149    fn with<R>(&self, f: impl FnOnce(&mut ScopeState) -> R) -> R {
150        f(&mut self.0.lock().unwrap_or_else(PoisonError::into_inner))
151    }
152
153    fn of(parts: &mut Parts) -> Self {
154        if let Some(scope) = parts.extensions.get::<Self>() {
155            return scope.clone();
156        }
157        let scope = Self::default();
158        parts.extensions.insert(scope.clone());
159        scope
160    }
161
162    /// Run registered teardowns (LIFO) on a spawned task.
163    fn spawn_teardowns(&self) {
164        let list = self.with(|s| std::mem::take(&mut s.teardowns));
165        if list.is_empty() {
166            return;
167        }
168        let Ok(runtime) = tokio::runtime::Handle::try_current() else {
169            tracing::warn!("no tokio runtime available; dependency teardown skipped");
170            return;
171        };
172        runtime.spawn(async move {
173            for teardown in list.into_iter().rev() {
174                if let Err(err) = tokio::spawn(teardown).await {
175                    tracing::error!(error = %err, "dependency teardown panicked");
176                }
177            }
178        });
179    }
180}
181
182type OverrideFn =
183    Arc<dyn Fn(RequestHead) -> BoxFut<'static, Result<Erased, ApiError>> + Send + Sync>;
184type GlobalFn =
185    Arc<dyn for<'a> Fn(&'a mut Parts) -> BoxFut<'a, Result<(), ApiError>> + Send + Sync>;
186
187/// Application-level DI configuration collected on an [`App`].
188#[derive(Default)]
189pub(crate) struct DiRegistry {
190    provided: HashMap<TypeId, Erased>,
191    overrides: HashMap<TypeId, OverrideFn>,
192    globals: Vec<GlobalFn>,
193}
194
195impl DiRegistry {
196    pub(crate) fn is_empty(&self) -> bool {
197        self.provided.is_empty() && self.overrides.is_empty() && self.globals.is_empty()
198    }
199}
200
201/// Chain of registries active for a request (innermost first).
202#[derive(Clone)]
203struct Active(Arc<ActiveNode>);
204
205struct ActiveNode {
206    registry: Arc<DiRegistry>,
207    parent: Option<Active>,
208}
209
210impl Active {
211    fn find<R>(&self, f: impl Fn(&DiRegistry) -> Option<R>) -> Option<R> {
212        let mut node = Some(&self.0);
213        while let Some(n) = node {
214            if let Some(found) = f(&n.registry) {
215                return Some(found);
216            }
217            node = n.parent.as_ref().map(|p| &p.0);
218        }
219        None
220    }
221}
222
223fn downcast<U: Send + Sync + 'static>(value: Erased) -> Result<Arc<U>, ApiError> {
224    value.downcast::<U>().map_err(|_| {
225        ApiError::internal(format!(
226            "dependency type mismatch for `{}`",
227            type_name::<U>()
228        ))
229    })
230}
231
232/// Chain of type names if resolving `tid` now would close a cycle.
233fn cycle_chain(state: &ScopeState, tid: TypeId, name: &'static str) -> Option<Vec<&'static str>> {
234    let start = state.stack.iter().position(|(t, _)| *t == tid)?;
235    let mut chain: Vec<&'static str> = state.stack[start..].iter().map(|(_, n)| *n).collect();
236    chain.push(name);
237    Some(chain)
238}
239
240async fn resolve_arc<U: Dependency>(parts: &mut Parts) -> Result<Arc<U>, ApiError> {
241    let tid = TypeId::of::<U>();
242    let name = type_name::<U>();
243    let scope = RequestScope::of(parts);
244    if let Some(hit) = scope.with(|s| s.cache.get(&tid).cloned()) {
245        return downcast(hit);
246    }
247    if let Some(chain) = scope.with(|s| cycle_chain(s, tid, name)) {
248        return Err(ApiError::internal(DependencyError::Cycle { chain }));
249    }
250    let active = parts.extensions.get::<Active>().cloned();
251    if let Some(value) = active
252        .as_ref()
253        .and_then(|a| a.find(|r| r.provided.get(&tid).cloned()))
254    {
255        return downcast(value);
256    }
257    let override_fn = active
258        .as_ref()
259        .and_then(|a| a.find(|r| r.overrides.get(&tid).cloned()));
260    scope.with(|s| s.stack.push((tid, name)));
261    let outcome = match override_fn {
262        Some(f) => f(RequestHead::from(&*parts)).await,
263        None => {
264            let mut ctx = ResolveContext { parts };
265            U::resolve(&mut ctx).await.map(|v| Arc::new(v) as Erased)
266        }
267    };
268    scope.with(|s| s.stack.pop());
269    let value = outcome?;
270    scope.with(|s| s.cache.insert(tid, Arc::clone(&value)));
271    downcast(value)
272}
273
274/// Passed to [`Dependency::resolve`]: request access plus nested resolution.
275pub struct ResolveContext<'a> {
276    parts: &'a mut Parts,
277}
278
279impl<'a> ResolveContext<'a> {
280    /// The request head (method, URI, headers, extensions). Other extractors
281    /// can be run against it: `Header::from_request_parts(ctx.parts())`.
282    pub fn parts(&mut self) -> &mut Parts {
283        self.parts
284    }
285
286    /// Resolve another dependency (cached per request, cycle-checked).
287    pub fn resolve<U: Dependency>(&mut self) -> BoxFut<'_, Result<Arc<U>, ApiError>> {
288        Box::pin(resolve_arc::<U>(self.parts))
289    }
290
291    /// Register async cleanup to run after the response is produced (LIFO).
292    /// See the [module docs](self) for the exact guarantees.
293    pub fn on_teardown<F>(&mut self, teardown: F)
294    where
295        F: Future + Send + 'static,
296        F::Output: TeardownOutput,
297    {
298        let wrapped: Teardown = Box::pin(async move {
299            if let Some(message) = teardown.await.failure() {
300                tracing::error!(error = %message, "dependency teardown failed");
301            }
302        });
303        RequestScope::of(self.parts).with(|s| s.teardowns.push(wrapped));
304    }
305}
306
307/// Extractor injecting a [`Dependency`] (request-scoped, cached).
308#[derive(Debug)]
309pub struct Depends<T>(pub Arc<T>);
310
311impl<T> Clone for Depends<T> {
312    fn clone(&self) -> Self {
313        Self(Arc::clone(&self.0))
314    }
315}
316
317impl<T> Deref for Depends<T> {
318    type Target = T;
319
320    fn deref(&self) -> &T {
321        &self.0
322    }
323}
324
325thread_local! {
326    static DESCRIBING: RefCell<Vec<TypeId>> = const { RefCell::new(Vec::new()) };
327}
328
329impl<T: Dependency> FromRequestParts for Depends<T> {
330    async fn from_request_parts(parts: &mut Parts) -> Result<Self, ApiError> {
331        resolve_arc::<T>(parts).await.map(Depends)
332    }
333
334    fn describe(op: &mut Operation, registry: &mut SchemaRegistry) {
335        let tid = TypeId::of::<T>();
336        let entered = DESCRIBING.with(|d| {
337            let mut stack = d.borrow_mut();
338            if stack.contains(&tid) {
339                false
340            } else {
341                stack.push(tid);
342                true
343            }
344        });
345        if entered {
346            T::describe(op, registry);
347            DESCRIBING.with(|d| d.borrow_mut().pop());
348        }
349    }
350}
351
352/// Extractor for a value registered with [`App::provide`] (any
353/// `Send + Sync + 'static` type, not only dependencies).
354#[derive(Debug)]
355pub struct Provided<T>(pub Arc<T>);
356
357impl<T> Clone for Provided<T> {
358    fn clone(&self) -> Self {
359        Self(Arc::clone(&self.0))
360    }
361}
362
363impl<T> Deref for Provided<T> {
364    type Target = T;
365
366    fn deref(&self) -> &T {
367        &self.0
368    }
369}
370
371impl<T: Send + Sync + 'static> FromRequestParts for Provided<T> {
372    async fn from_request_parts(parts: &mut Parts) -> Result<Self, ApiError> {
373        let tid = TypeId::of::<T>();
374        let found = parts
375            .extensions
376            .get::<Active>()
377            .and_then(|a| a.find(|r| r.provided.get(&tid).cloned()));
378        match found {
379            Some(value) => downcast(value).map(Provided),
380            None => Err(ApiError::internal(DependencyError::NotProvided {
381                type_name: type_name::<T>(),
382            })),
383        }
384    }
385}
386
387impl App {
388    /// Share `value` with every request of this app (and its mounts). See the
389    /// [module docs](self).
390    #[must_use]
391    pub fn provide<T: Send + Sync + 'static>(self, value: T) -> Self {
392        self.provide_arc(Arc::new(value))
393    }
394
395    /// Like [`App::provide`] for a value that is already shared, so the
396    /// caller can keep a handle to it.
397    #[must_use]
398    pub fn provide_arc<T: Send + Sync + 'static>(mut self, value: Arc<T>) -> Self {
399        self.di.provided.insert(TypeId::of::<T>(), value);
400        self
401    }
402
403    /// Replace `T::resolve` with `f` (also for nested resolution). `f` gets an
404    /// owned [`RequestHead`] because a borrowed request cannot cross the
405    /// returned future.
406    #[must_use]
407    pub fn override_dependency<T, F, Fut>(mut self, f: F) -> Self
408    where
409        T: Dependency,
410        F: Fn(RequestHead) -> Fut + Send + Sync + 'static,
411        Fut: Future<Output = Result<T, ApiError>> + Send + 'static,
412    {
413        let erased: OverrideFn = Arc::new(move |head| {
414            let fut = f(head);
415            Box::pin(async move { fut.await.map(|v| Arc::new(v) as Erased) })
416        });
417        self.di.overrides.insert(TypeId::of::<T>(), erased);
418        self
419    }
420
421    /// Replace `T::resolve` with a clone of `value`.
422    #[must_use]
423    pub fn override_value<T: Dependency + Clone>(self, value: T) -> Self {
424        self.override_dependency(move |_| {
425            let value = value.clone();
426            async move { Ok(value) }
427        })
428    }
429
430    /// Resolve `T` before every routed request; its error short-circuits.
431    /// The value stays cached, so handlers asking for `Depends<T>` reuse it.
432    #[must_use]
433    pub fn dependency<T: Dependency>(mut self) -> Self {
434        self.di.globals.push(Arc::new(|parts: &mut Parts| {
435            Box::pin(async move { resolve_arc::<T>(parts).await.map(|_| ()) })
436        }));
437        self
438    }
439}
440
441/// Wrap `router` so requests carry the DI scope, run global dependencies and
442/// trigger teardown.
443pub(crate) fn install(router: axum::Router, registry: DiRegistry) -> axum::Router {
444    router.layer(DiLayer(Arc::new(registry)))
445}
446
447#[derive(Clone)]
448struct DiLayer(Arc<DiRegistry>);
449
450impl<S> Layer<S> for DiLayer {
451    type Service = DiService<S>;
452
453    fn layer(&self, inner: S) -> DiService<S> {
454        DiService {
455            inner,
456            registry: Arc::clone(&self.0),
457        }
458    }
459}
460
461#[derive(Clone)]
462struct DiService<S> {
463    inner: S,
464    registry: Arc<DiRegistry>,
465}
466
467impl<S> Service<axum::extract::Request> for DiService<S>
468where
469    S: Service<axum::extract::Request, Response = axum::response::Response, Error = Infallible>
470        + Clone
471        + Send
472        + 'static,
473    S::Future: Send,
474{
475    type Response = axum::response::Response;
476    type Error = Infallible;
477    type Future = BoxFut<'static, Result<Self::Response, Infallible>>;
478
479    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Infallible>> {
480        self.inner.poll_ready(cx)
481    }
482
483    fn call(&mut self, req: axum::extract::Request) -> Self::Future {
484        let ready = self.inner.clone();
485        let inner = std::mem::replace(&mut self.inner, ready);
486        let registry = Arc::clone(&self.registry);
487        Box::pin(async move {
488            let (mut parts, body) = req.into_parts();
489            crate::middleware::note_matched_path(&parts);
490            let parent = parts.extensions.get::<Active>().cloned();
491            let node = ActiveNode {
492                registry: Arc::clone(&registry),
493                parent,
494            };
495            parts.extensions.insert(Active(Arc::new(node)));
496            let owned = match parts.extensions.get::<RequestScope>() {
497                Some(_) => None,
498                None => {
499                    let scope = RequestScope::default();
500                    parts.extensions.insert(scope.clone());
501                    Some(scope)
502                }
503            };
504            let response = match run_globals(&registry, &mut parts).await {
505                Err(err) => err.into_response().map(Body::into_inner),
506                Ok(()) => {
507                    inner
508                        .oneshot(http::Request::from_parts(parts, body))
509                        .await?
510                }
511            };
512            if let Some(scope) = owned {
513                scope.spawn_teardowns();
514            }
515            Ok(response)
516        })
517    }
518}
519
520async fn run_globals(registry: &DiRegistry, parts: &mut Parts) -> Result<(), ApiError> {
521    for global in &registry.globals {
522        global(parts).await?;
523    }
524    Ok(())
525}
siderite_core/di.rs

siderite_core/
di.rs
