Axum End to End

#[derive(Clone)] struct AppState { db: sqlx::PgPool }
#[derive(serde::Deserialize)] struct CreateTransfer { amount_kobo: i64, to: String }

async fn create(State(s): State<AppState>, headers: HeaderMap, Json(body): Json<CreateTransfer>)
    -> Result<(StatusCode, Json<TransferView>), ApiError>
{
    let key = headers.get("idempotency-key").ok_or(ApiError::MissingKey)?.to_str().map_err(|_| ApiError::MissingKey)?;
    if body.amount_kobo <= 0 { return Err(ApiError::Validation("amount_kobo must be positive")); }
    let t = transfers::create(&s.db, &body, key).await?;
    Ok((StatusCode::CREATED, Json(t)))
}

impl IntoResponse for ApiError {                      // one problem-details format for every error
    fn into_response(self) -> Response {
        let (status, code) = match &self { ApiError::Insufficient => (StatusCode::UNPROCESSABLE_ENTITY, "insufficient_funds"),
            ApiError::MissingKey => (StatusCode::BAD_REQUEST, "missing_idempotency_key"), _ => (StatusCode::INTERNAL_SERVER_ERROR, "internal") };
        (status, Json(json!({ "type": code, "detail": self.to_string() }))).into_response()
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let db = PgPoolOptions::new().max_connections(20).connect(&std::env::var("DATABASE_URL")?).await?;
    sqlx::migrate!().run(&db).await?;
    let app = Router::new().route("/v1/transfers", post(create)).with_state(AppState { db })
        .layer(TraceLayer::new_for_http()).layer(TimeoutLayer::new(Duration::from_secs(10)));
    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
    axum::serve(listener, app).with_graceful_shutdown(shutdown_signal()).await?;
    Ok(())
}

// sqlx: checked against the schema at compile time (DATABASE_URL or offline metadata)
let row = sqlx::query!("UPDATE accounts SET balance_kobo = balance_kobo - $1 WHERE id = $2 AND balance_kobo >= $1 RETURNING balance_kobo", amt, id)
    .fetch_optional(&mut *tx).await?;