Rust and Axum source walkthrough

The runnable source below is included from the same files packaged in the notes download. Lockfiles, configuration, and the shared browser UI are included in the download.

Database

A notes request reaches this adapter after input validation. DATABASE_ENGINE chooses PostgreSQL, MariaDB, or MongoDB; DATABASE_URL comes from the app’s attachment. SQL uses parameters rather than interpolating note text, and MongoDB uses driver operations. The initialization process creates a table or unique index. Records live in the database, so replacing a web instance leaves them available to its replacement.

#![allow(unused)]
fn main() {
enum Database {
    Postgres(PgPool),
    Maria(MySqlPool),
    Mongo(Collection<Document>),
}
static DATABASE: OnceCell<Database> = OnceCell::const_new();
async fn database() -> Result<&'static Database> {
    DATABASE
        .get_or_try_init(|| async {
            let url = std::env::var("DATABASE_URL").context("Attach a database first")?;
            Ok(match env("DATABASE_ENGINE", "postgresql").as_str() {
                "postgresql" => Database::Postgres(
                    sqlx::postgres::PgPoolOptions::new()
                        .max_connections(5)
                        .connect(&url)
                        .await?,
                ),
                "mariadb" => Database::Maria(
                    sqlx::mysql::MySqlPoolOptions::new()
                        .max_connections(5)
                        .connect(&url)
                        .await?,
                ),
                "mongodb" => {
                    let client = mongodb::Client::with_uri_str(&url).await?;
                    Database::Mongo(
                        client
                            .default_database()
                            .context("DATABASE_URL needs a database name")?
                            .collection("notes"),
                    )
                }
                _ => anyhow::bail!("Unsupported DATABASE_ENGINE"),
            })
        })
        .await
}
pub async fn migrate() -> Result<()> {
    let query = "CREATE TABLE IF NOT EXISTS notes (id VARCHAR(36) PRIMARY KEY, body TEXT NOT NULL)";
    match database().await? {
        Database::Postgres(db) => {
            sqlx::query(query).execute(db).await?;
        }
        Database::Maria(db) => {
            sqlx::query(query).execute(db).await?;
        }
        Database::Mongo(db) => {
            db.create_index(
                IndexModel::builder()
                    .keys(doc! {"id": 1})
                    .options(IndexOptions::builder().unique(true).build())
                    .build(),
            )
            .await?;
        }
    }
    Ok(())
}
pub async fn notes() -> Result<Vec<Note>> {
    Ok(match database().await? {
        Database::Postgres(db) => sqlx::query("SELECT id, body FROM notes ORDER BY id")
            .fetch_all(db)
            .await?
            .into_iter()
            .map(|r| Note {
                id: r.get("id"),
                body: r.get("body"),
            })
            .collect(),
        Database::Maria(db) => sqlx::query("SELECT id, body FROM notes ORDER BY id")
            .fetch_all(db)
            .await?
            .into_iter()
            .map(|r| Note {
                id: r.get("id"),
                body: r.get("body"),
            })
            .collect(),
        Database::Mongo(db) => db
            .find(doc! {})
            .sort(doc! {"id":1})
            .await?
            .try_collect::<Vec<_>>()
            .await?
            .into_iter()
            .map(|d| {
                Ok(Note {
                    id: d.get_str("id")?.into(),
                    body: d.get_str("body")?.into(),
                })
            })
            .collect::<Result<Vec<_>>>()?,
    })
}
pub async fn add_note(body: String) -> Result<Note> {
    let note = Note {
        id: uuid::Uuid::new_v4().to_string(),
        body,
    };
    match database().await? {
        Database::Postgres(db) => {
            sqlx::query("INSERT INTO notes (id, body) VALUES ($1, $2)")
                .bind(&note.id)
                .bind(&note.body)
                .execute(db)
                .await?;
        }
        Database::Maria(db) => {
            sqlx::query("INSERT INTO notes (id, body) VALUES (?, ?)")
                .bind(&note.id)
                .bind(&note.body)
                .execute(db)
                .await?;
        }
        Database::Mongo(db) => {
            db.insert_one(doc! {"id":&note.id,"body":&note.body})
                .await?;
        }
    }
    Ok(note)
}
}

Worker

Creating an export enqueues an ID and returns immediately. This loop reads the notes, writes an export under that ID, records its status, and acknowledges the stream entry. It also reclaims jobs abandoned before acknowledgement. Retrying writes the same export key, so interruption does not create an additional export file; the snapshot can reflect newer notes if they changed before the retry. See worker recovery.

#![allow(unused)]
fn main() {
async fn process(q: &mut Queue, message: Message) -> Result<()> {
    let (message_id, fields) = message;
    let id = fields
        .chunks_exact(2)
        .find(|pair| pair[0] == "id")
        .map(|pair| pair[1].clone())
        .unwrap_or_default();
    if uuid::Uuid::parse_str(&id).is_ok() {
        if status(q, &id).await?.as_deref() != Some("ready") {
            save_export(q, &id, serde_json::to_vec_pretty(&notes().await?)?).await?;
            redis::cmd("SET")
                .arg(format!("export:{id}"))
                .arg("ready")
                .query_async::<()>(q)
                .await?;
        }
        println!("Export ready: {id}");
    }
    redis::cmd("XACK")
        .arg("exports")
        .arg("workers")
        .arg(&message_id)
        .query_async::<i64>(q)
        .await?;
    redis::cmd("XDEL")
        .arg("exports")
        .arg(message_id)
        .query_async::<i64>(q)
        .await?;
    Ok(())
}
async fn next(q: &mut Queue, consumer: &str) -> Result<()> {
    let (_, mut messages, _): (String, Vec<Message>, Vec<String>) = redis::cmd("XAUTOCLAIM")
        .arg("exports")
        .arg("workers")
        .arg(consumer)
        .arg(30000)
        .arg("0-0")
        .arg("COUNT")
        .arg(1)
        .query_async(q)
        .await?;
    if messages.is_empty() {
        let reply: ReadReply = redis::cmd("XREADGROUP")
            .arg("GROUP")
            .arg("workers")
            .arg(consumer)
            .arg("COUNT")
            .arg(1)
            .arg("BLOCK")
            .arg(5000)
            .arg("STREAMS")
            .arg("exports")
            .arg(">")
            .query_async(q)
            .await?;
        messages = reply
            .and_then(|mut streams| streams.pop().map(|(_, messages)| messages))
            .unwrap_or_default();
    }
    for message in messages {
        process(q, message).await?;
    }
    Ok(())
}
async fn run() -> Result<()> {
    let mut q = queue().await?;
    let group = redis::cmd("XGROUP")
        .arg("CREATE")
        .arg("exports")
        .arg("workers")
        .arg("0")
        .arg("MKSTREAM")
        .query_async::<()>(&mut q)
        .await;
    if let Err(error) = group {
        if !error.to_string().contains("BUSYGROUP") {
            return Err(error.into());
        }
    }
    let consumer = uuid::Uuid::new_v4().to_string();
    println!("Worker ready");
    let stop = shutdown();
    tokio::pin!(stop);
    loop {
        tokio::select! {
            _ = &mut stop => return Ok(()),
            result = next(&mut q, &consumer) => {
                if result.is_err() {
                    eprintln!("Export failed; unacknowledged jobs retry after 30 seconds. Check attachments and schema.");
                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                    if let Ok(connection) = queue().await { q = connection; }
                }
            }
        }
    }
}
}

Object storage

Without S3 configuration, exports are stored in Redis for the worker checkpoint. With an attached bucket, the worker writes JSON to exports/<id>.json and the web process reads that object for downloads. Both processes need their own bucket attachment. The client uses the supplied endpoint and region with path-style requests. Credentials stay on the server; a private bucket does not add authentication to the example’s public download route.

#![allow(unused)]
fn main() {
async fn storage() -> Result<aws_sdk_s3::Client> {
    let shared = aws_config::defaults(aws_config::BehaviorVersion::latest())
        .region(aws_config::Region::new(std::env::var("S3_REGION")?))
        .load()
        .await;
    let config = aws_sdk_s3::config::Builder::from(&shared)
        .endpoint_url(std::env::var("S3_ENDPOINT")?)
        .force_path_style(true)
        .build();
    Ok(aws_sdk_s3::Client::from_conf(config))
}
pub async fn save_export(q: &mut Queue, id: &str, data: Vec<u8>) -> Result<()> {
    if let Ok(bucket) = std::env::var("S3_BUCKET") {
        storage()
            .await?
            .put_object()
            .bucket(bucket)
            .key(format!("exports/{id}.json"))
            .content_type("application/json")
            .body(aws_sdk_s3::primitives::ByteStream::from(data))
            .send()
            .await?;
    } else {
        redis::cmd("SET")
            .arg(format!("export:{id}:data"))
            .arg(data)
            .query_async::<()>(q)
            .await?;
    }
    Ok(())
}
pub async fn load_export(q: &mut Queue, id: &str) -> Result<Vec<u8>> {
    if let Ok(bucket) = std::env::var("S3_BUCKET") {
        let output = storage()
            .await?
            .get_object()
            .bucket(bucket)
            .key(format!("exports/{id}.json"))
            .send()
            .await?;
        Ok(output.body.collect().await?.into_bytes().to_vec())
    } else {
        Ok(redis::cmd("GET")
            .arg(format!("export:{id}:data"))
            .query_async(q)
            .await?)
    }
}
}

HTTP routes

use axum::{
    Json, Router,
    extract::Path,
    http::{StatusCode, header},
    response::{Html, IntoResponse, Response},
    routing::{get, post},
};
use d2o_notes::*;
use serde_json::{Value, json};

struct Unavailable;
impl From<anyhow::Error> for Unavailable {
    fn from(_: anyhow::Error) -> Self {
        Self
    }
}
impl IntoResponse for Unavailable {
    fn into_response(self) -> Response {
        (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({"error":"Check attachments, schema, and service availability"})),
        )
            .into_response()
    }
}
async fn list() -> Result<Json<Vec<Note>>, Unavailable> {
    Ok(Json(notes().await?))
}
async fn create(Json(input): Json<Value>) -> Result<Response, Unavailable> {
    let Some(body) = input
        .get("body")
        .and_then(Value::as_str)
        .filter(|s| !s.trim().is_empty() && s.chars().count() <= 2000)
    else {
        return Ok((
            StatusCode::BAD_REQUEST,
            Json(json!({"error":"Enter a note of 1–2000 characters"})),
        )
            .into_response());
    };
    Ok((StatusCode::CREATED, Json(add_note(body.into()).await?)).into_response())
}
async fn export() -> Result<Response, Unavailable> {
    Ok((StatusCode::ACCEPTED, Json(json!({"id":enqueue().await?}))).into_response())
}
async fn export_status(Path(id): Path<String>) -> Result<Response, Unavailable> {
    let state = status(&mut queue().await?, &id).await?;
    Ok((
        if state.is_some() {
            StatusCode::OK
        } else {
            StatusCode::NOT_FOUND
        },
        Json(json!({"status":state.unwrap_or("unknown".into())})),
    )
        .into_response())
}
async fn download(Path(id): Path<String>) -> Result<Response, Unavailable> {
    let mut q = queue().await?;
    if uuid::Uuid::parse_str(&id).is_err() || status(&mut q, &id).await?.as_deref() != Some("ready")
    {
        return Ok((
            StatusCode::NOT_FOUND,
            Json(json!({"error":"Export not ready"})),
        )
            .into_response());
    }
    Ok((
        [
            (header::CONTENT_TYPE, "application/json"),
            (
                header::CONTENT_DISPOSITION,
                "attachment; filename=notes.json",
            ),
        ],
        load_export(&mut q, &id).await?,
    )
        .into_response())
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let app = Router::new()
        .route(
            "/",
            get(|| async { Html(include_str!("../public/index.html")) }),
        )
        .route("/health", get(|| async { "ok\n" }))
        .route("/notes", get(list).post(create))
        .route("/exports", post(export))
        .route("/exports/{id}", get(export_status))
        .route("/exports/{id}/download", get(download))
        .layer(axum::extract::DefaultBodyLimit::max(16384));
    let listener =
        tokio::net::TcpListener::bind(format!("0.0.0.0:{}", env("PORT", "3000"))).await?;
    println!("Web ready");
    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown())
        .await?;
    Ok(())
}

Review production preparation before adapting these handlers to real users.