Pipes

Stateless transforms at the request boundary — parse, normalize, validate before the handler runs.

A pipe runs between extraction and the handler. The transport pulls a value out of the request — a path segment, a query param, a JSON body, a GraphQL argument, a WS payload, an MCP tool’s arguments, a queued job — and hands it to the pipe. The pipe either returns the transformed value, or rejects with a PipeError the transport renders as its native error.

The trait lives in nest-rs-pipes, transport-agnostic, and binds per argument on every transport. Two forms, one surface:

HTTP wraps an extractor: nest_rs::http::Piped<P, E> / Valid<E> run the pipe over what E extracts.
GraphQL, WebSockets, MCP, and queues deserialize a single typed value, so there is no extractor to wrap: nest_rs::pipes::Piped<P, T> / Valid<T> expose the wire type T and hand the handler the transformed value. The transport macro (#[operations], #[messages], #[tools], #[processor]) does the stripping and the call. On MCP the carrier goes inside Parameters<…>, since that is what the protocol deserializes an operation’s arguments into.

pub trait Pipe {
    type In;
    type Out;
    fn transform(input: Self::In) -> Result<Self::Out, PipeError>;
}

Install

cargo add nest-rs --features pipes

The http feature turns this on, and http is a default — name it explicitly only in a headless app built with default-features = false, where no transport implies it.

Pipes are stateless

A pipe is a zero-sized marker named at a call site — never instantiated, never injected. transform is an associated function; there’s no &self and no DI container in scope. That’s deliberate: a pipe is a pure transform, the same input always produces the same output, and the type name is enough to pick which one runs.

use nest_rs::pipes::{Pipe, PipeError};

pub struct Trim;

impl Pipe for Trim {
    type In = String;
    type Out = String;
    fn transform(input: String) -> Result<String, PipeError> {
        Ok(input.trim().to_string())
    }
}

If you need DI-injected logic at the request boundary, that’s a service or an interceptor — not a pipe.

The built-in set

nest-rs-pipes ships the common cases. Reach for them before writing your own:

Pipe	In → Out	Use it for
`Parse<T>`	`String → T` (any `FromStr`)	Generic conversion: `ParseInt`, `ParseFloat`, `ParseBool`
`ParseUuid`	`String → Uuid`	Any-version UUID
`ParseUuidV4` / `ParseUuidV7` / …	`String → Uuid`	Version-pinned UUID
`ParseArray<P>`	`String → Vec<P::Out>`	Comma-separated list, each item piped through `P`
`Trim`, `Lowercase`, `Uppercase`	`String → String`	Edge normalization
`ValidationPipe<T>`	`T → T` (where `T: Validate`)	Run validator attribute rules

A PipeError rejection surfaces as the transport’s native error — the handler body never runs:

Transport	A rejection becomes
HTTP	RFC 9457 `problem+json` `400` (`{ "type": …, "title": "Bad Request", "status": 400, "detail": "must be a UUID v7" }`)
GraphQL	An error in the response’s `errors` array
WebSockets	An error frame on the socket (`{ event, data: { error, errors } }`)
MCP	`invalid_params` (`-32602`), the field errors in the error’s `data`
Queue	A job error — the queue’s retry/failure policy applies

ValidationPipe<T> attaches the structured field-level errors to the rejection, and every transport carries them through under the name errors: an RFC 9457 extension member on HTTP, extensions.errors on GraphQL, data.errors on a WebSocket frame, the error data on MCP, an errors field on the queue’s job dead-lettered event. A client (or an operator reading a dead-letter log) can always tell which field was wrong, not merely that something was.

Validation at the HTTP edge

Valid<E> is the ergonomic form of Piped<ValidationPipe<T>, E>: extract E, then validate.

use nest_rs::http::poem::web::Json;
use nest_rs::http::{Valid, input};

#[input]
pub struct CreateUser {
    #[validate(email)]
    email: String,
    #[validate(length(min = 8))]
    password: String,
}

#[post("/")]
async fn create(&self, Valid(input): Valid<Json<CreateUser>>)
    -> Result<Json<User>>
{
    Ok(Json(self.svc.create(input).await?))
}

A malformed JSON body is 400 before the pipe runs. A well-formed body that fails validation is an RFC 9457 problem+json 400 with:

{
  "type": "https://www.rfc-editor.org/rfc/rfc9110#status.400",
  "title": "Bad Request",
  "status": 400,
  "detail": "validation failed",
  "errors": {
    "email": [{ "code": "email", "message": null, "params": { /* ... */ } }],
    "password": [{ "code": "length", "params": { "min": 8 } }]
  }
}

No manual checks in the handler.

Custom transforms with `Piped<P, E>`

When the input needs a transform other than Validate, name the pipe at the call site:

use nest_rs::http::Piped;
use nest_rs::pipes::ParseUuidV7;
use nest_rs::http::poem::web::Path;

#[get("/:id")]
async fn get(&self, id: Piped<ParseUuidV7, Path<String>>) -> Json<User> {
    Json(self.svc.find(*id).await?)
}

Valid<E> destructures in the parameter list (Valid(input)) and binds the extractor’s inner value — Valid<Json<CreateUser>> carries the CreateUser, never the Json. Piped<P, E> names its pipe with a phantom marker, so it is not a tuple struct: bind it and reach the value through Deref/into_inner(), as above.

Path<String> extracts the raw segment; ParseUuidV7::transform validates it as a v7 UUID. A non-UUID string is 400 with "must be a UUID v7" — not a stringly-typed handler doing its own parse::<Uuid>().

The same pipe beyond HTTP

On GraphQL, WebSockets, MCP and queues the argument is the wire value, so the binding is nest_rs::pipes::Piped<P, T> / Valid<T>. T is what appears on the wire — the SDL argument type, the WS payload, the job payload — and the handler body sees P::Out through into_inner() / Deref, exactly as on HTTP.

// GraphQL — the SDL argument is `String`; the body sees it trimmed.
#[query]
#[public]
async fn trimmed(&self, raw: Piped<Trim, String>) -> async_graphql::Result<String> {
    Ok(raw.into_inner())
}

// GraphQL — validates the input object; the SDL argument is `NameInput`.
#[query]
#[public]
async fn named(&self, input: Valid<NameInput>) -> async_graphql::Result<String> {
    Ok(input.into_inner().name)
}

// WebSockets — the message payload is a `String`, piped before the handler.
#[subscribe_message("trim")]
#[public]
async fn trim_handler(&self, name: Piped<Trim, String>) -> String {
    name.into_inner()
}

// MCP — the tool's schema is `Coords`; the body sees it validated.
#[tool]
#[public]
async fn current_weather(
    &self,
    Parameters(coords): Parameters<Valid<Coords>>,
) -> Result<String, McpError> {
    let coords = coords.into_inner();
    /* ... */
}

// Queue — the job payload is a `String`, piped after deserialization.
#[process(queue = SignupsQueue, retries = 0)]
async fn handle(&self, name: Piped<Trim, String>) -> anyhow::Result<()> {
    /* ... */
}

Pipes vs `Bind`

A pipe converts a primitive input — an ID string into a Uuid, a JSON blob into a validated DTO. A Bind<A, S> extractor goes further: it parses the id, loads the row through a service, and authorizes it — returning 404 if the row doesn’t exist within the caller’s scope, 403 if denied.

// Pipe — pure conversion, no DB
#[get("/_validate/:id")]
async fn validate(&self, id: Piped<ParseUuidV7, Path<String>>) -> String {
    format!("{} is valid", *id)
}

// Bind — convert + load + authorize
#[get("/:id")]
async fn get(&self, user: Bind<Read, UsersService>) -> Json<User> {
    Json(User::from(&*user))
}

Bind depends on the data layer (its GraphQL analog is bind — see By-id binding). Pipe is transport-agnostic — the same ParseUuidV7 binds identically on HTTP, GraphQL, WebSockets, and queues.

Writing a pipe

Three rules:

Stateless. transform is an associated function; no &self, no container.
One file per pipe under crates/nest-rs-pipes/src/pipes/ (one role, one file). Reusable pipes belong to the framework crate — never to an app.
Use the error envelope. Return PipeError::new(msg) for a simple message, PipeError::with_details(msg, details) to carry structured field-level errors.

use nest_rs::pipes::{Pipe, PipeError};

pub struct NonEmpty;

impl Pipe for NonEmpty {
    type In = String;
    type Out = String;
    fn transform(input: String) -> Result<String, PipeError> {
        if input.trim().is_empty() {
            Err(PipeError::new("must not be empty"))
        } else {
            Ok(input)
        }
    }
}

Then bind it at the call site:

#[post("/")]
async fn create(&self, name: Piped<NonEmpty, Json<String>>) -> &'static str {
    "ok"
}

Going further

HTTP / controllers — every extractor that composes with a pipe.
Guards — gate access; pipes don’t.
Database / Bind — pipes are for primitives, Bind loads and authorizes a row.
oauth/http/controller.rs in the demo — binds Piped<Lowercase, Path<String>> so the OAuth provider segment is case-insensitive.