Requests
Fans can browse concerts. Now they want to search for their favourite artists, and buy tickets. The staff want to upload posters, and send whole line-ups over HTTP. Margaret reads request data into your own structs, and validates it before your route runs.
Query strings
Describe what you expect with serde and validator, both with their derive features:
use serde::Deserialize;
use validator::Validate;
#[derive(Deserialize, Validate)]
pub struct SearchForm {
#[validate(length(min = 2))]
pub artist: String,
}
Then ask for it with #[form_request]:
#[singleton]
#[responds_to_http(method = RouteMethod::Get, path = "/concerts/search", server = "public")]
pub struct SearchConcerts {
concerts: Arc<ConcertRepository>,
}
impl SearchConcerts {
#[constructor]
pub fn create(concerts: Arc<ConcertRepository>) -> anyhow::Result<Self> {
Ok(Self { concerts })
}
#[process]
pub async fn respond(
&self,
#[form_request(from = RequestInput::Query)] SearchForm { artist }: SearchForm,
) -> anyhow::Result<Response> {
let lines: Vec<String> = self
.concerts
.by_artist(artist)
.await?
.into_iter()
.map(|ConcertWithVenue { concert, venue }| {
format!(
"{} {} at {}, {}",
concert.starts_at.format("%b %e"),
concert.artist,
venue.name,
venue.city
)
})
.collect();
Ok(Response::text(200, lines.join("\n")))
}
}
Your route only runs with valid data. Search for a one-letter artist, and Margaret answers with a 422 and the validation errors, all by itself. Leave out artist altogether, and it answers 400. RequestInput also has Form for HTML forms, Cookie for cookies, and Json.
Query strings, forms, and cookies carry text, so give their fields types that read from text, like String or Uuid. Numbers and lists belong in JSON requests.
Searching by an index
You saw in Database that filtering concerts by artist doesn't compile without an index. Searching by artist is now a real need, so let's add one. It lists each artist's concerts in the order they start:
#[model(table = "concerts")]
#[index(name = "concerts_by_start", fields = [starts_at, id])]
#[index(name = "concerts_by_artist", fields = [artist, starts_at, id])]
#[derive(Clone)]
pub struct Concert {
Now the query walks it: the artist first, then the start time from now on:
pub async fn by_artist(
&self,
artist: String,
) -> Result<Vec<ConcertWithVenue>, ActiveRecordError> {
Ok(Concert::query()
.artist
.eq(artist)
.starts_at
.above(self.clock.now())
.ascending()
.limit::<20>()
.load::<ConcertWithVenue, _>(self.database.as_ref())
.await?
.records)
}
Recreate the database with box_office schema to get the new index.
JSON requests
Buying tickets happens in the seat picker, which sends the order as JSON. Orders need customers, orders, and tickets, so here are three new models. A fieldless enum works as a column, too:
#[derive(Clone, Copy, PartialEq)]
pub enum OrderStatus {
Reserved,
Paid,
}
#[model(table = "orders")]
#[index(name = "orders_by_customer", fields = [customer, id])]
#[index(name = "orders_by_concert", fields = [concert, id])]
#[derive(Clone)]
pub struct Order {
#[column(primary_key, default = ColumnDefault::UuidV7)]
pub id: Uuid,
#[column]
#[foreign_key(on_delete = OnDelete::Cascade)]
pub customer: Key<Customer>,
#[column]
#[foreign_key(on_delete = OnDelete::Restrict)]
pub concert: Key<Concert>,
#[column]
pub status: OrderStatus,
#[column]
pub reserved_until: DateTime<Utc>,
}
A seat can only be sold once per concert. Constraints that span several fields go on the struct:
#[model(table = "tickets")]
#[unique(fields = [concert, seat])]
#[index(name = "tickets_by_order", fields = [order, id])]
#[derive(Clone)]
pub struct Ticket {
#[column(primary_key, default = ColumnDefault::UuidV7)]
pub id: Uuid,
#[column]
#[foreign_key(on_delete = OnDelete::Cascade)]
pub order: Key<Order>,
#[column]
#[foreign_key(on_delete = OnDelete::Restrict)]
pub concert: Key<Concert>,
#[column(minimum = 1)]
pub seat: i32,
}
A primary key can span several fields too, with #[primary_key(fields = [concert, seat])] on the struct. Tickets keep an id of their own instead, because routes bind models by a single-field primary key, and the door scanner will look tickets up by id.
Customer has a primary key and a unique email. The checkout request itself is a plain struct:
use serde::Deserialize;
use validator::Validate;
#[derive(Deserialize, Validate)]
pub struct CheckoutRequest {
#[validate(email)]
pub email: String,
#[validate(length(min = 1, max = 8))]
pub seats: Vec<i32>,
}
A route that reads a request body declares how big that body may be, with max_body_bytes. A GET route can't read a body at all. The seat picker wants to show its own error messages, so this route asks for a ValidationResult and decides what to answer:
#[singleton]
#[responds_to_http(
max_body_bytes = 16_384,
method = RouteMethod::Post,
name = "place_order",
path = "/concerts/{concert}/orders",
server = "public"
)]
pub struct PlaceOrder {
orders: Arc<OrderRepository>,
}
impl PlaceOrder {
#[constructor]
pub fn create(orders: Arc<OrderRepository>) -> anyhow::Result<Self> {
Ok(Self { orders })
}
#[process]
pub async fn respond(
&self,
#[route_parameter(from = "concert")] concert: Concert,
#[form_request(from = RequestInput::Json)] checkout: ValidationResult<CheckoutRequest>,
) -> anyhow::Result<Response> {
let CheckoutRequest { email, seats } = match checkout {
ValidationResult::Valid(checkout) => checkout,
ValidationResult::Invalid(errors) => return Ok(Response::json(422, &errors)),
ValidationResult::Malformed(malformation) => {
return Ok(Response::json(
400,
&json!({ "error": malformation.to_string() }),
));
}
};
if seats.iter().any(|seat| *seat > concert.capacity) {
return Ok(Response::json(
422,
&json!({ "error": format!("Pick seats from 1 to {}.", concert.capacity) }),
));
}
Ok(match self.orders.reserve(&concert, email, &seats).await? {
Reservation::Reserved { order_id } => {
Response::json(201, &json!({ "order_id": order_id }))
}
Reservation::SalesClosed => Response::json(
409,
&json!({ "error": "This concert has already started." }),
),
Reservation::SeatsTaken => Response::json(
409,
&json!({ "error": "Someone was faster. Please pick other seats." }),
),
})
}
}
A body that isn't JSON at all never reaches the route: Margaret answers 400, 415 for the wrong content type, and 413 when it's too big.
Reserving seats in a transaction
The OrderRepository reserves the seats in one transaction, so a customer either gets all their seats or none:
pub async fn reserve(
&self,
concert: &Concert,
email: String,
seats: &[i32],
) -> anyhow::Result<Reservation> {
let now = self.clock.now();
let mut connection = self.database.connection().await?;
let transaction = connection.transaction(Isolation::ReadCommitted).await?;
let customer = match Customer::query()
.email
.eq(email.clone())
.find(&transaction)
.await?
{
Lookup::Found(customer) => customer,
Lookup::Missing => {
Customer::create(CustomerDraft { email })
.run(&transaction)
.await?
}
};
let order = match Order::create(OrderDraft {
customer: Key::of(&customer),
concert: Key::of(concert),
status: OrderStatus::Reserved,
reserved_until: now + TimeDelta::minutes(15),
})
.when(
Concert::query()
.id
.eq(concert.id)
.when(|concert| concert.starts_at.above(now))
.exists::<Detached>(),
)
.run(&transaction)
.await?
{
Creation::Created(order) => order,
Creation::Refused => {
transaction.rollback().await?;
return Ok(Reservation::SalesClosed);
}
};
for seat in seats {
match Ticket::create(TicketDraft {
order: Key::of(&order),
concert: Key::of(concert),
seat: *seat,
})
.run(&transaction)
.await
{
Ok(_) => {}
Err(ActiveRecordError::UniqueViolation { .. }) => {
transaction.rollback().await?;
return Ok(Reservation::SeatsTaken);
}
Err(error) => return Err(error.into()),
}
}
transaction.commit().await?;
Ok(Reservation::Reserved { order_id: order.id })
}
Two guarantees come from the database itself:
.when(...) guards the order: it's created only if the concert still hasn't started at that very moment, otherwise create answers Creation::Refused.
When someone else was faster, the unique constraint on [concert, seat] answers with ActiveRecordError::UniqueViolation, and the customer gets a friendly 409.
$ curl -H 'content-type: application/json' -d '{"email":"ana@example.com","seats":[12,13]}' \
http://127.0.0.1:8080/concerts/01a13257-80cf-7996-ad23-ed86b5b49159/orders
{"order_id":"01a1325b-39ae-7ed4-9716-c6bc635cd4cf"}
$ curl -H 'content-type: application/json' -d '{"email":"bo@example.com","seats":[13,14]}' \
http://127.0.0.1:8080/concerts/01a13257-80cf-7996-ad23-ed86b5b49159/orders
{"error":"Someone was faster. Please pick other seats."}
File uploads
Staff upload concert posters on the admin server. A poster belongs to exactly one concert, so its primary key is the concert's key:
use margaret::framework::active_record::key::Key;
use margaret::framework::macros::model;
use margaret::framework::model::on_delete::OnDelete;
use crate::concerts::concert::Concert;
#[model(table = "posters")]
pub struct Poster {
#[column(primary_key)]
#[foreign_key(on_delete = OnDelete::Cascade)]
pub concert: Key<Concert>,
#[column]
pub content_type: String,
#[column]
pub image: Vec<u8>,
}
A route asks for the uploaded files with an UploadedFiles parameter. Uploading a new poster replaces the old one, so the route inserts the poster, and updates the existing row when there's one already:
#[singleton]
#[responds_to_http(
max_body_bytes = 5_242_880,
method = RouteMethod::Post,
path = "/concerts/{concert}/poster",
server = "admin"
)]
pub struct UploadPoster {
database: Arc<Database>,
}
impl UploadPoster {
#[constructor]
pub fn create(database: Arc<Database>) -> anyhow::Result<Self> {
Ok(Self { database })
}
#[process]
pub async fn respond(
&self,
#[route_parameter(from = "concert")] concert: Concert,
files: UploadedFiles,
) -> anyhow::Result<Response> {
let Some(upload) = files.get("poster") else {
return Ok(Response::text(
422,
"Attach the poster as a \"poster\" file.",
));
};
let poster = Poster {
concert: Key::of(&concert),
content_type: upload.content_type().to_string(),
image: tokio::fs::read(upload.path()).await?,
};
poster
.insert()
.or_update(|columns| {
columns
.content_type
.excluded()
.and(columns.image.excluded())
})
.run(self.database.as_ref())
.await?;
Ok(Response::text(
201,
format!("Saved the poster for {}.", concert.artist),
))
}
}
excluded() means "the value you tried to insert". Each upload is a temporary file, removed when the request ends, so the route reads it before answering. Uploads need a directory, so serve now asks for one, but only for the servers that accept uploads:
$ box_office serve --admin-upload-dir /var/lib/box_office/uploads ...
Streaming request bodies
Promoters' booking systems want to push whole seasons over HTTP, in the same JSON Lines format the import command reads. A season can be large, so instead of buffering the body, the route asks for a RequestBodyStream and imports each concert as soon as its line arrives:
#[singleton]
#[responds_to_http(
max_body_bytes = 10_485_760,
method = RouteMethod::Put,
path = "/concerts/import",
server = "admin"
)]
pub struct StreamConcertImport {
database: Arc<Database>,
importer: Arc<ConcertImporter>,
}
impl StreamConcertImport {
#[constructor]
pub fn create(database: Arc<Database>, importer: Arc<ConcertImporter>) -> anyhow::Result<Self> {
Ok(Self { database, importer })
}
#[process]
pub async fn respond(&self, mut body: RequestBodyStream) -> anyhow::Result<Response> {
let mut connection = self.database.connection().await?;
let transaction = connection.transaction(Isolation::ReadCommitted).await?;
let mut pending = Vec::new();
let mut imported = 0;
loop {
let finished = match body.next_chunk().await {
RequestBodyChunk::Data(chunk) => {
pending.extend_from_slice(&chunk);
false
}
RequestBodyChunk::End => {
pending.push(b'\n');
true
}
RequestBodyChunk::Rejected(rejection) => return Ok(rejection.into_response()),
};
while let Some(end) = pending.iter().position(|byte| *byte == b'\n') {
let line: Vec<u8> = pending.drain(..=end).collect();
if line.trim_ascii().is_empty() {
continue;
}
let listing: ConcertListing = match serde_json::from_slice(&line) {
Ok(listing) => listing,
Err(error) => {
return Ok(Response::text(
422,
format!("Line {} isn't a concert: {error}", imported + 1),
));
}
};
self.importer.import(&transaction, &listing).await?;
imported += 1;
}
if finished {
transaction.commit().await?;
return Ok(Response::text(
200,
format!("Imported {imported} concerts."),
));
}
}
}
}
Each chunk is Data, End, or Rejected, for example when the body grows past max_body_bytes. The same ConcertImporter from the command does the work, inside one transaction again: a broken line rolls back the whole season.
$ curl -T season.jsonl http://10.0.0.5:9090/concerts/import
Imported 2 concerts.