📁 Rust Backend with Axum · Part 37 of 42

Rust Web Server with Axum 0.8: CI with GitHub Actions, and Deploy

November 3, 2026

Episode 37 of the Rust Backend series, where we build the backend of a blog step by step. After episode 36 we have tests and a Docker image. But will we really run them before every push? Today a robot does it for us: GitHub Actions. And then it builds our image. All the code below is the final, tested code from the video, ready to copy.

New to Rust? Start with the beginner series.
The full backend series playlist is here.

We continue in the same blog_api project from episode 36. You need a GitHub account and the gh command line tool (gh auth login once). The full project, after cargo fmt, is at the end of this post.

1. Continuous integration

CI (continuous integration): on every push, a fresh machine checks out the code and runs our checks. If one fails, the commit gets a red cross.

text

push → fresh machine → cargo fmt --check → cargo clippy → cargo test

2. A repository

bash

git init -q -b main && git add . && git commit -q -m 'Blog API after episode 36' && cat .gitignore
gh repo create axum-blog-api --public --source . --remote origin

.gitignore keeps .env, the target folder and the database out: secrets never go into git.

3. The workflow

A workflow is a YAML file in .github/workflows/. Ours runs on every push to main, and on pull requests:

yaml

name: CI
on:
  push:
    branches: [main]
  pull_request:

jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v5
      - uses: dtolnay/rust-toolchain@stable
        with:
          components: rustfmt, clippy
      - uses: Swatinem/rust-cache@v2
      - run: cargo fmt --check
      - run: cargo clippy --all-targets -- -D warnings
      - run: cargo test

actions/checkout gets the code; dtolnay/rust-toolchain installs Rust with rustfmt and clippy.
Swatinem/rust-cache keeps target and the downloaded crates between runs, so later runs are much faster.
Three commands: formatting must already be done, clippy treats every warning as an error, and the tests must pass.

bash

git add . && git commit -q -m 'Add CI' && git push -u origin main 2>&1 | tail -3
gh run watch

4. Red: cargo fmt

Red, after a few seconds:

bash

gh run view --log-failed $ID | cut -f3 | cut -c30- | tail -14

cargo fmt --check found code that is not formatted: about 80 places. We never ran cargo fmt. It rewrites every file in the standard style:

bash

cargo fmt && git commit -q -am 'cargo fmt' && git push -q

5. Red: clippy

Red again, now in clippy. Warnings are errors now:

bash

gh run view --log-failed $ID | cut -f3 | cut -c30- | grep -B3 -A7 -m1 'is never read'

text

error: field `0` is never read

It is AdminUser(pub AuthUser) from episode 21: the extractor carries the user, but nobody ever looks at it. Clippy is right. We use it: the log says who listed the users.

rust

pub async fn list_users<R: Repo>(
    State(repo): State<R>,
    AdminUser(admin): AdminUser,
) -> Result<Json<Vec<User>>, ApiError> {
    tracing::info!(admin = admin.id, "listing users");
    Ok(Json(repo.list_users().await?))
}

And we run clippy on our own machine, to be sure:

bash

cargo clippy --all-targets -- -D warnings
git commit -q -am 'Use the admin' && git push -q

6. Red: the tests need JWT_SECRET

Now the tests fail:

bash

gh run view --log-failed $ID | cut -f3 | cut -c30- | grep -m4 -E 'JWT_SECRET|test result'

text

JWT_SECRET must be set

Two of five tests fail. On our machine the value comes from .env, which is not in git. The tests need a value, any value, so the workflow sets one:

yaml

  check:
    runs-on: ubuntu-latest
    env:
      JWT_SECRET: test-only-secret

A real secret would go into the repository's secrets (Settings → Secrets and variables → Actions) and be read as ${{ secrets.NAME }}, never written into the file.

bash

git commit -q -am 'Set JWT_SECRET for the tests' && git push -q
gh run watch

Green. Every step passed.

7. Build and push the image

A second job builds the Docker image and pushes it to GitHub's container registry, ghcr.io. But only on the main branch, and only if the checks passed:

yaml

  image:
    needs: check
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v5
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - uses: docker/build-push-action@v6
        with:
          push: true
          tags: ghcr.io/${{ github.repository }}:latest

needs: check: wait for the first job, and skip this one if it failed.
It logs in with GITHUB_TOKEN, which every workflow run gets automatically. permissions: packages: write lets it push images, and nothing else.

bash

git commit -q -am 'Build and push the image' && git push -q
gh run list -L 5

Both jobs run; the image build takes a few minutes. Then the image is on GitHub, tagged latest. A public repository gives a public package, so anyone can pull it without a login.

8. Deploy

Deploy means: a server pulls the new image and runs it. Our server is this machine. Redis first, then our image from GitHub; docker run pulls it for us:

bash

docker network create blog && docker run -d --name redis --network blog redis:7
docker run -d --name blog --network blog -p 8080:8080 -v blog-data:/data -e JWT_SECRET=5c0e8f1b3a9d47e2b6f0a4c8d1e7239f6b5a0c3d8e1f47a29b6c5d0e3f8a1b47 -e REDIS_URL=redis://redis:6379 ghcr.io/mahbd/axum-blog-api:latest
curl -i localhost:8080/health/ready

text

HTTP/1.1 200 OK

{"database":"up","redis":"up"}

This is the image that the robot built. (Replace mahbd with your GitHub name, and use your own secret.)

9. A badge

A badge in the README shows the status to everyone:

markdown

# axum-blog-api

![CI](https://github.com/mahbd/axum-blog-api/actions/workflows/ci.yml/badge.svg)

The blog API from the series "Rust Backend with Axum".

bash

git add . && git commit -q -m 'Add a README with a badge' && git push -q && gh repo view | head -12

The final code

.github/workflows/ci.yml

yaml

name: CI
on:
  push:
    branches: [main]
  pull_request:

jobs:
  check:
    runs-on: ubuntu-latest
    env:
      JWT_SECRET: test-only-secret
    steps:
      - uses: actions/checkout@v5
      - uses: dtolnay/rust-toolchain@stable
        with:
          components: rustfmt, clippy
      - uses: Swatinem/rust-cache@v2
      - run: cargo fmt --check
      - run: cargo clippy --all-targets -- -D warnings
      - run: cargo test

  image:
    needs: check
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v5
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - uses: docker/build-push-action@v6
        with:
          push: true
          tags: ghcr.io/${{ github.repository }}:latest

README.md

markdown

# axum-blog-api

![CI](https://github.com/mahbd/axum-blog-api/actions/workflows/ci.yml/badge.svg)

The blog API from the series "Rust Backend with Axum".

src/api_tests.rs

rust

use axum::body::{Body, to_bytes};
use axum::extract::ConnectInfo;
use axum::http::{Method, Request, StatusCode, header};
use serde_json::{Value, json};
use sqlx::SqlitePool;
use tower::ServiceExt;

use super::*;

fn test_app(pool: SqlitePool) -> Router {
    let (jobs, _rx) = mpsc::channel(10);
    let (live, _) = broadcast::channel(10);
    let client = redis::Client::open("redis://127.0.0.1:1").unwrap();
    let config = ConnectionManagerConfig::new()
        .set_number_of_retries(0)
        .set_connection_timeout(Some(Duration::from_millis(100)));
    let cache = ConnectionManager::new_lazy_with_config(client, config).unwrap();
    let handle = PrometheusBuilder::new().build_recorder().handle();
    app(
        SqliteRepo { pool },
        MemStorage::default(),
        jobs,
        cache,
        live,
        handle,
    )
}

async fn call(
    app: &Router,
    method: Method,
    uri: &str,
    token: Option<&str>,
    body: Value,
) -> (StatusCode, Value) {
    let mut req = Request::builder()
        .method(method)
        .uri(uri)
        .header(header::CONTENT_TYPE, "application/json");
    if let Some(t) = token {
        req = req.header(header::AUTHORIZATION, format!("Bearer {t}"));
    }
    let mut req = req.body(Body::from(body.to_string())).unwrap();
    req.extensions_mut()
        .insert(ConnectInfo(SocketAddr::from(([127, 0, 0, 1], 4000))));
    let res = app.clone().oneshot(req).await.unwrap();
    let status = res.status();
    let bytes = to_bytes(res.into_body(), usize::MAX).await.unwrap();
    (
        status,
        serde_json::from_slice(&bytes).unwrap_or(Value::Null),
    )
}

async fn login(app: &Router, email: &str) -> String {
    let creds = json!({ "email": email, "password": "hunter2hunter2" });
    call(app, Method::POST, "/signup", None, creds.clone()).await;
    let (status, body) = call(app, Method::POST, "/login", None, creds).await;
    assert_eq!(status, StatusCode::OK);
    body["token"].as_str().unwrap().to_string()
}

#[sqlx::test]
async fn create_and_get(pool: SqlitePool) {
    let app = test_app(pool);
    let token = login(&app, "ana@example.com").await;
    let new = json!({ "title": "Hello", "body": "First post" });
    let (status, created) = call(&app, Method::POST, "/posts", Some(&token), new).await;
    assert_eq!(status, StatusCode::CREATED);
    let (status, post) = call(
        &app,
        Method::GET,
        &format!("/posts/{}", created["id"]),
        None,
        Value::Null,
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(post["title"], "Hello");
}

#[sqlx::test]
async fn ben_cannot_delete_ana_post(pool: SqlitePool) {
    let app = test_app(pool);
    let ana = login(&app, "ana@example.com").await;
    let ben = login(&app, "ben@example.com").await;
    let new = json!({ "title": "Mine", "body": "Hands off" });
    let (_, post) = call(&app, Method::POST, "/posts", Some(&ana), new).await;
    let uri = format!("/posts/{}", post["id"]);
    let (status, _) = call(&app, Method::DELETE, &uri, Some(&ben), Value::Null).await;
    assert_eq!(status, StatusCode::FORBIDDEN);
    let (status, _) = call(&app, Method::DELETE, &uri, Some(&ana), Value::Null).await;
    assert_eq!(status, StatusCode::NO_CONTENT);
}

#[sqlx::test]
async fn database_down_gives_500(pool: SqlitePool) {
    let app = test_app(pool.clone());
    pool.close().await;
    let (status, body) = call(&app, Method::GET, "/posts/1", None, Value::Null).await;
    assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
    assert_eq!(body["error"], "database error");
}

src/auth.rs

rust

use argon2::password_hash::{PasswordVerifier, phc::PasswordHash};
use argon2::{Argon2, password_hash::PasswordHasher};
use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation, decode, encode};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::time::{SystemTime, UNIX_EPOCH};

use crate::error::ApiError;

pub async fn hash_password(password: String) -> Result<String, ApiError> {
    let hash = tokio::task::spawn_blocking(move || {
        Argon2::default()
            .hash_password(password.as_bytes())
            .map(|h| h.to_string())
    })
    .await
    .expect("hash task panicked")?;
    Ok(hash)
}

#[derive(Serialize, Deserialize)]
pub struct Claims {
    pub sub: i64,
    pub role: String,
    pub exp: u64,
}

pub fn make_token(user_id: i64, role: &str, secret: &str) -> Result<String, ApiError> {
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs();
    let claims = Claims {
        sub: user_id,
        role: role.to_string(),
        exp: now + 900,
    };
    Ok(encode(
        &Header::default(),
        &claims,
        &EncodingKey::from_secret(secret.as_bytes()),
    )?)
}

pub async fn verify_password(password: String, hash: String) -> Result<bool, ApiError> {
    let ok = tokio::task::spawn_blocking(move || {
        let parsed = PasswordHash::new(&hash)?;
        Ok::<_, argon2::password_hash::Error>(
            Argon2::default()
                .verify_password(password.as_bytes(), &parsed)
                .is_ok(),
        )
    })
    .await
    .expect("verify task panicked")?;
    Ok(ok)
}

pub fn read_token(token: &str, secret: &str) -> Result<Claims, ApiError> {
    let mut validation = Validation::new(Algorithm::HS256);
    validation.leeway = 0;
    let data = decode::<Claims>(
        token,
        &DecodingKey::from_secret(secret.as_bytes()),
        &validation,
    )
    .map_err(|err| {
        tracing::info!(?err, "bad token");
        ApiError::Unauthorized
    })?;
    Ok(data.claims)
}

pub const REFRESH_SECS: i64 = 30 * 24 * 3600;

pub fn now() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64
}

pub fn new_refresh_token() -> String {
    let mut bytes = [0u8; 32];
    rand::fill(&mut bytes);
    hex::encode(bytes)
}

pub fn hash_token(token: &str) -> String {
    hex::encode(Sha256::digest(token.as_bytes()))
}

src/cache.rs

rust

use redis::{AsyncCommands, aio::ConnectionManager};

use crate::models::Post;

pub async fn get(redis: &ConnectionManager, id: i64) -> Option<Post> {
    let mut conn = redis.clone();
    match conn.get::<_, Option<String>>(format!("post:{id}")).await {
        Ok(Some(json)) => serde_json::from_str(&json).ok(),
        Ok(None) => None,
        Err(err) => {
            tracing::warn!(?err, "redis get failed");
            None
        }
    }
}

pub async fn put(redis: &ConnectionManager, post: &Post) {
    let mut conn = redis.clone();
    let json = serde_json::to_string(post).unwrap();
    if let Err(err) = conn
        .set_ex::<_, _, ()>(format!("post:{}", post.id), json, 60)
        .await
    {
        tracing::warn!(?err, "redis set failed");
    }
}

pub async fn del(redis: &ConnectionManager, id: i64) {
    let mut conn = redis.clone();
    if let Err(err) = conn.del::<_, ()>(format!("post:{id}")).await {
        tracing::warn!(?err, "redis del failed");
    }
}

src/config.rs

rust

use std::sync::OnceLock;

pub struct Config {
    pub host: String,
    pub port: u16,
    pub database_url: String,
    pub jwt_secret: String,
    pub storage: String,
    pub smtp_host: String,
    pub smtp_port: u16,
    pub mail_from: String,
    pub cleanup_secs: u64,
    pub redis_url: String,
}

impl Config {
    fn from_env() -> Config {
        dotenvy::dotenv().ok();
        let host = std::env::var("HOST").unwrap_or("0.0.0.0".to_string());
        let port = std::env::var("PORT").unwrap_or("3000".to_string());
        let port = port
            .parse()
            .unwrap_or_else(|e| panic!("PORT {port:?} is not a port number: {e}"));
        let database_url = std::env::var("DATABASE_URL").unwrap_or("sqlite://blog.db".to_string());
        let jwt_secret = std::env::var("JWT_SECRET").expect("JWT_SECRET must be set");
        let storage = std::env::var("STORAGE").unwrap_or("local".to_string());
        let smtp_host = std::env::var("SMTP_HOST").unwrap_or("localhost".to_string());
        let smtp_port = std::env::var("SMTP_PORT")
            .unwrap_or("1025".to_string())
            .parse()
            .expect("SMTP_PORT is a port number");
        let mail_from =
            std::env::var("MAIL_FROM").unwrap_or("LCE Blog <blog@example.com>".to_string());
        let cleanup_secs = std::env::var("CLEANUP_SECS")
            .unwrap_or("3600".to_string())
            .parse()
            .expect("CLEANUP_SECS is a number");
        let redis_url = std::env::var("REDIS_URL").unwrap_or("redis://127.0.0.1:6379".to_string());
        Config {
            host,
            port,
            database_url,
            jwt_secret,
            storage,
            smtp_host,
            smtp_port,
            mail_from,
            cleanup_secs,
            redis_url,
        }
    }
}

static CONFIG: OnceLock<Config> = OnceLock::new();

pub fn config() -> &'static Config {
    CONFIG.get_or_init(Config::from_env)
}

src/error.rs

rust

use axum::Json;
use axum::extract::multipart::MultipartError;
use axum::extract::rejection::{JsonRejection, QueryRejection};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use serde::Serialize;
use serde_json::json;
use std::collections::BTreeMap;
use thiserror::Error;
use utoipa::ToSchema;

#[derive(Debug, Error)]
pub enum ApiError {
    #[error("post {0} not found")]
    NotFound(i64),
    #[error(transparent)]
    BadJson(#[from] JsonRejection),
    #[error(transparent)]
    BadQuery(#[from] QueryRejection),
    #[error("database error")]
    Database(#[from] sqlx::Error),
    #[error("{0}")]
    Conflict(String),
    #[error("validation failed")]
    Validation(#[from] validator::ValidationErrors),
    #[error("internal error")]
    Hash(#[from] argon2::password_hash::Error),
    #[error("internal error")]
    Jwt(#[from] jsonwebtoken::errors::Error),
    #[error("invalid email or password")]
    Unauthorized,
    #[error("you are not allowed to do this")]
    Forbidden,
    #[error("file not found")]
    FileNotFound,
    #[error("{0}")]
    Multipart(#[from] MultipartError),
    #[error("send a file in the form field file")]
    MissingFile,
    #[error("internal error")]
    Io(#[from] std::io::Error),
    #[error("only PNG and JPEG images")]
    UnsupportedMedia,
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        tracing::warn!(error = %self, "request failed");
        let status = match &self {
            ApiError::NotFound(_) => StatusCode::NOT_FOUND,
            ApiError::Conflict(_) => StatusCode::CONFLICT,
            ApiError::Unauthorized => StatusCode::UNAUTHORIZED,
            ApiError::Forbidden => StatusCode::FORBIDDEN,
            ApiError::FileNotFound => StatusCode::NOT_FOUND,
            ApiError::UnsupportedMedia => StatusCode::UNSUPPORTED_MEDIA_TYPE,
            ApiError::MissingFile => StatusCode::BAD_REQUEST,
            ApiError::Multipart(err) => err.status(),
            ApiError::BadJson(rejection) => rejection.status(),
            ApiError::BadQuery(rejection) => rejection.status(),
            ApiError::Validation(errors) => {
                let fields: BTreeMap<_, Vec<String>> = errors
                    .field_errors()
                    .iter()
                    .map(|(field, errs)| {
                        let msgs = errs
                            .iter()
                            .map(|e| e.message.clone().unwrap_or(e.code.clone()).to_string())
                            .collect();
                        (field.to_string(), msgs)
                    })
                    .collect();
                return (
                    StatusCode::UNPROCESSABLE_ENTITY,
                    Json(json!({ "error": "validation failed", "fields": fields })),
                )
                    .into_response();
            }
            ApiError::Hash(err) => {
                tracing::error!(?err, "password hashing failed");
                StatusCode::INTERNAL_SERVER_ERROR
            }
            ApiError::Jwt(err) => {
                tracing::error!(?err, "token error");
                StatusCode::INTERNAL_SERVER_ERROR
            }
            ApiError::Database(err) => {
                tracing::error!(?err, "database error");
                StatusCode::INTERNAL_SERVER_ERROR
            }
            ApiError::Io(err) => {
                tracing::error!(?err, "io error");
                StatusCode::INTERNAL_SERVER_ERROR
            }
        };
        let message = match &self {
            ApiError::BadJson(rejection) => rejection.body_text(),
            ApiError::BadQuery(rejection) => rejection.body_text(),
            ApiError::Multipart(err) => err.body_text(),
            other => other.to_string(),
        };
        (status, Json(json!({ "error": message }))).into_response()
    }
}

#[derive(Serialize, ToSchema)]
pub struct ErrorBody {
    pub error: String,
}

src/extract.rs

rust

use axum::Json;
use axum::extract::{FromRequest, FromRequestParts, Request};
use axum::http::{header, request::Parts};
use serde::de::DeserializeOwned;
use std::str::FromStr;
use validator::Validate;

use crate::auth;
use crate::config::config;
use crate::error::ApiError;

pub struct ValidJson<T>(pub T);

impl<T, S> FromRequest<S> for ValidJson<T>
where
    T: DeserializeOwned + Validate,
    S: Send + Sync,
{
    type Rejection = ApiError;

    async fn from_request(req: Request, state: &S) -> Result<Self, Self::Rejection> {
        let Json(value) = Json::<T>::from_request(req, state).await?;
        value.validate()?;
        Ok(ValidJson(value))
    }
}

pub struct AuthUser {
    pub id: i64,
    pub role: Role,
}

impl<S: Send + Sync> FromRequestParts<S> for AuthUser {
    type Rejection = ApiError;

    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
        let token = parts
            .headers
            .get(header::AUTHORIZATION)
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "))
            .ok_or(ApiError::Unauthorized)?;
        let claims = auth::read_token(token, &config().jwt_secret)?;
        Ok(AuthUser {
            id: claims.sub,
            role: claims.role.parse()?,
        })
    }
}

#[derive(Clone, Copy, PartialEq, Debug)]
pub enum Role {
    User,
    Admin,
}

impl FromStr for Role {
    type Err = ApiError;

    fn from_str(s: &str) -> Result<Role, ApiError> {
        match s {
            "user" => Ok(Role::User),
            "admin" => Ok(Role::Admin),
            _ => Err(ApiError::Unauthorized),
        }
    }
}

pub struct AdminUser(pub AuthUser);

impl<S: Send + Sync> FromRequestParts<S> for AdminUser {
    type Rejection = ApiError;

    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
        let user = AuthUser::from_request_parts(parts, state).await?;
        if user.role != Role::Admin {
            return Err(ApiError::Forbidden);
        }
        Ok(AdminUser(user))
    }
}

src/handlers.rs

rust

use axum::extract::{Extension, Multipart, Path, Query, State, rejection::QueryRejection};
use axum::{
    Json,
    http::{StatusCode, header},
    response::{IntoResponse, Response},
};

use crate::auth;
use crate::cache;
use crate::config::config;
use crate::error::{ApiError, ErrorBody};
use crate::extract::{AdminUser, AuthUser, Role, ValidJson};
use crate::jobs::Job;
use crate::models::{
    Comment, Hit, Image, ListFilter, ListParams, NewComment, NewPost, NewUser, Page, Post, PostV2,
    PostWithTags, SearchParams, User,
};
use crate::repo::Repo;
use crate::storage::Storage;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum_extra::extract::cookie::{Cookie, CookieJar, SameSite};
use metrics_exporter_prometheus::PrometheusHandle;
use redis::aio::ConnectionManager;
use serde::Deserialize;
use serde_json::{Value, json};
use std::convert::Infallible;
use std::io;
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::mpsc;
use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};
use uuid::Uuid;

#[utoipa::path(
    post,
    path = "/posts",
    request_body = NewPost,
    responses(
        (status = 201, description = "Created", body = PostWithTags),
        (status = 401, description = "Missing or bad token", body = ErrorBody),
        (status = 422, description = "Validation failed", body = ErrorBody),
    ),
    security(("bearer" = [])),
    tag = "posts"
)]
#[tracing::instrument(skip_all)]
pub async fn create_post<R: Repo>(
    State(repo): State<R>,
    user: AuthUser,
    ValidJson(new): ValidJson<NewPost>,
) -> Result<(StatusCode, Json<PostWithTags>), ApiError> {
    let post = repo
        .create(user.id, new)
        .await
        .map_err(|err| match err.as_database_error() {
            Some(e) if e.is_unique_violation() => {
                ApiError::Conflict("a post with this title already exists".into())
            }
            _ => ApiError::Database(err),
        })?;
    tracing::debug!(id = post.post.id, "created");
    metrics::counter!("posts_created_total").increment(1);
    Ok((StatusCode::CREATED, Json(post)))
}

pub async fn get_post<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
    Path(id): Path<i64>,
) -> Result<Json<Post>, ApiError> {
    if let Some(post) = cache::get(&redis, id).await {
        tracing::debug!(id, "cache hit");
        return Ok(Json(post));
    }
    tracing::debug!(id, "cache miss");
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    cache::put(&redis, &post).await;
    Ok(Json(post))
}

pub async fn list_posts<R: Repo>(
    State(repo): State<R>,
    params: Result<Query<ListParams>, QueryRejection>,
) -> Result<Json<Page<Post>>, ApiError> {
    let Query(p) = params?;
    let page = p.page.unwrap_or(1).max(1);
    let per_page = p.per_page.unwrap_or(10).clamp(1, 100);
    let (items, total) = repo
        .list(ListFilter {
            page,
            per_page,
            tag: p.tag,
            sort: p.sort,
        })
        .await?;
    Ok(Json(Page {
        items,
        page,
        per_page,
        total,
    }))
}

pub async fn update_post<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
    Path(id): Path<i64>,
    user: AuthUser,
    ValidJson(new): ValidJson<NewPost>,
) -> Result<Json<Post>, ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    if !can_edit(&user, &post) {
        return Err(ApiError::Forbidden);
    }
    let post = repo.update(id, new).await?.ok_or(ApiError::NotFound(id))?;
    cache::del(&redis, id).await;
    Ok(Json(post))
}

pub async fn delete_post<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
    Path(id): Path<i64>,
    user: AuthUser,
) -> Result<StatusCode, ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    if !can_edit(&user, &post) {
        return Err(ApiError::Forbidden);
    }
    repo.delete(id).await?;
    cache::del(&redis, id).await;
    Ok(StatusCode::NO_CONTENT)
}

pub async fn list_comments<R: Repo>(
    State(repo): State<R>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<Comment>>, ApiError> {
    repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    Ok(Json(repo.list_comments(id).await?))
}

pub async fn add_comment<R: Repo>(
    State(repo): State<R>,
    Path(id): Path<i64>,
    Extension(live): Extension<broadcast::Sender<Comment>>,
    _user: AuthUser,
    ValidJson(new): ValidJson<NewComment>,
) -> Result<(StatusCode, Json<Comment>), ApiError> {
    let comment = repo
        .add_comment(id, new)
        .await
        .map_err(|err| match err.as_database_error() {
            Some(db_err) if db_err.is_foreign_key_violation() => ApiError::NotFound(id),
            _ => ApiError::Database(err),
        })?;
    let _ = live.send(comment.clone());
    Ok((StatusCode::CREATED, Json(comment)))
}

pub async fn search_posts<R: Repo>(
    State(repo): State<R>,
    Query(p): Query<SearchParams>,
) -> Result<Json<Vec<Hit>>, ApiError> {
    Ok(Json(repo.search(&p.q).await?))
}

pub async fn restore_post<R: Repo>(
    State(repo): State<R>,
    Path(id): Path<i64>,
) -> Result<StatusCode, ApiError> {
    if !repo.restore(id).await? {
        return Err(ApiError::NotFound(id));
    }
    Ok(StatusCode::OK)
}

pub async fn signup<R: Repo>(
    State(repo): State<R>,
    Extension(jobs): Extension<mpsc::Sender<Job>>,
    ValidJson(new): ValidJson<NewUser>,
) -> Result<(StatusCode, Json<User>), ApiError> {
    let hash = crate::auth::hash_password(new.password).await?;
    let user =
        repo.create_user(&new.email, &hash)
            .await
            .map_err(|err| match err.as_database_error() {
                Some(e) if e.is_unique_violation() => {
                    ApiError::Conflict("email already registered".into())
                }
                _ => ApiError::Database(err),
            })?;
    if jobs
        .send(Job::Welcome {
            email: user.email.clone(),
        })
        .await
        .is_err()
    {
        tracing::error!("job queue closed");
    }
    Ok((StatusCode::CREATED, Json(user)))
}

#[derive(Deserialize)]
pub struct Login {
    pub email: String,
    pub password: String,
}

fn refresh_cookie(token: String) -> Cookie<'static> {
    Cookie::build(("refresh", token))
        .http_only(true)
        .secure(true)
        .same_site(SameSite::Strict)
        .path("/auth")
        .max_age(time::Duration::days(30))
        .build()
}

pub async fn login<R: Repo>(
    State(repo): State<R>,
    jar: CookieJar,
    Json(body): Json<Login>,
) -> Result<(CookieJar, Json<Value>), ApiError> {
    let user = repo
        .find_user_by_email(&body.email)
        .await?
        .ok_or(ApiError::Unauthorized)?;
    if !auth::verify_password(body.password, user.password_hash).await? {
        return Err(ApiError::Unauthorized);
    }
    let access = auth::make_token(user.id, &user.role, &config().jwt_secret)?;
    let refresh = auth::new_refresh_token();
    repo.store_refresh(
        &auth::hash_token(&refresh),
        user.id,
        auth::now() + auth::REFRESH_SECS,
    )
    .await?;
    Ok((
        jar.add(refresh_cookie(refresh)),
        Json(json!({ "token": access })),
    ))
}

pub async fn me<R: Repo>(State(repo): State<R>, user: AuthUser) -> Result<Json<User>, ApiError> {
    let user = repo
        .get_user(user.id)
        .await?
        .ok_or(ApiError::Unauthorized)?;
    Ok(Json(user))
}

pub async fn refresh<R: Repo>(
    State(repo): State<R>,
    jar: CookieJar,
) -> Result<(CookieJar, Json<Value>), ApiError> {
    let old = jar
        .get("refresh")
        .ok_or(ApiError::Unauthorized)?
        .value()
        .to_string();
    let user_id = repo
        .take_refresh(&auth::hash_token(&old), auth::now())
        .await?
        .ok_or(ApiError::Unauthorized)?;
    let user = repo
        .get_user(user_id)
        .await?
        .ok_or(ApiError::Unauthorized)?;
    let access = auth::make_token(user.id, &user.role, &config().jwt_secret)?;
    let new = auth::new_refresh_token();
    repo.store_refresh(
        &auth::hash_token(&new),
        user.id,
        auth::now() + auth::REFRESH_SECS,
    )
    .await?;
    Ok((
        jar.add(refresh_cookie(new)),
        Json(json!({ "token": access })),
    ))
}

pub async fn logout<R: Repo>(
    State(repo): State<R>,
    jar: CookieJar,
) -> Result<(CookieJar, StatusCode), ApiError> {
    if let Some(cookie) = jar.get("refresh") {
        repo.delete_refresh(&auth::hash_token(cookie.value()))
            .await?;
    }
    Ok((
        jar.remove(Cookie::build("refresh").path("/auth")),
        StatusCode::NO_CONTENT,
    ))
}

pub fn can_edit(user: &AuthUser, post: &Post) -> bool {
    user.role == Role::Admin || post.author_id == Some(user.id)
}

pub async fn list_users<R: Repo>(
    State(repo): State<R>,
    AdminUser(admin): AdminUser,
) -> Result<Json<Vec<User>>, ApiError> {
    tracing::info!(admin = admin.id, "listing users");
    Ok(Json(repo.list_users().await?))
}

pub async fn upload_image<R: Repo, S: Storage>(
    State(repo): State<R>,
    user: AuthUser,
    Path(id): Path<i64>,
    Extension(store): Extension<Arc<S>>,
    mut multipart: Multipart,
) -> Result<(StatusCode, Json<Image>), ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    if !can_edit(&user, &post) {
        return Err(ApiError::Forbidden);
    }
    while let Some(field) = multipart.next_field().await? {
        if field.name() != Some("file") {
            continue;
        }
        let bytes = field.bytes().await?;
        let ext = image_kind(&bytes).ok_or(ApiError::UnsupportedMedia)?;
        let name = format!("{}.{ext}", Uuid::new_v4());
        store.put(&name, &bytes).await?;
        let image = repo.add_image(id, &format!("/files/{name}")).await?;
        return Ok((StatusCode::CREATED, Json(image)));
    }
    Err(ApiError::MissingFile)
}

fn image_kind(bytes: &[u8]) -> Option<&'static str> {
    if bytes.starts_with(&[0x89, b'P', b'N', b'G']) {
        Some("png")
    } else if bytes.starts_with(&[0xFF, 0xD8, 0xFF]) {
        Some("jpg")
    } else {
        None
    }
}

pub async fn get_file<S: Storage>(
    Extension(store): Extension<Arc<S>>,
    Path(name): Path<String>,
) -> Result<Response, ApiError> {
    let bytes = store.get(&name).await.map_err(|err| match err.kind() {
        io::ErrorKind::NotFound | io::ErrorKind::InvalidInput => ApiError::FileNotFound,
        _ => ApiError::Io(err),
    })?;
    let content_type = if name.ends_with(".png") {
        "image/png"
    } else {
        "image/jpeg"
    };
    Ok(([(header::CONTENT_TYPE, content_type)], bytes).into_response())
}

pub async fn live(
    Extension(live): Extension<broadcast::Sender<Comment>>,
    Path(post_id): Path<i64>,
    ws: WebSocketUpgrade,
) -> Response {
    let rx = live.subscribe();
    ws.on_upgrade(move |socket| live_socket(socket, rx, post_id))
}

async fn live_socket(mut socket: WebSocket, mut rx: broadcast::Receiver<Comment>, post_id: i64) {
    loop {
        tokio::select! {
            event = rx.recv() => match event {
                Ok(c) => {
                    if c.post_id != post_id {
                        continue;
                    }
                    let text = serde_json::to_string(&c).unwrap();
                    if socket.send(Message::Text(text.into())).await.is_err() {
                        break;
                    }
                }
                Err(RecvError::Lagged(n)) => {
                    let _ = socket.send(Message::Text(format!("{{\"missed\":{n}}}").into())).await;
                }
                Err(RecvError::Closed) => break,
            },
            msg = socket.recv() => match msg {
                None | Some(Ok(Message::Close(_))) | Some(Err(_)) => break,
                Some(Ok(_)) => {}
            },
        }
    }
    tracing::info!(post_id, "live socket closed");
}

pub async fn events(
    Extension(live): Extension<broadcast::Sender<Comment>>,
    Path(post_id): Path<i64>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let stream = BroadcastStream::new(live.subscribe()).filter_map(move |msg| match msg {
        Ok(c) if c.post_id == post_id => {
            Some(Ok(Event::default().event("comment").json_data(&c).unwrap()))
        }
        _ => None,
    });
    Sse::new(stream).keep_alive(KeepAlive::default())
}

#[utoipa::path(
    get,
    path = "/api/v2/posts/{id}",
    params(("id" = i64, Path, description = "Post id")),
    responses(
        (status = 200, description = "The post", body = PostV2),
        (status = 404, description = "No such post", body = ErrorBody),
    ),
    tag = "posts"
)]
pub async fn get_post_v2<R: Repo>(
    State(repo): State<R>,
    Path(id): Path<i64>,
) -> Result<Json<PostV2>, ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    Ok(Json(post.into()))
}

#[derive(Deserialize)]
pub struct SlowParams {
    secs: Option<u64>,
}

pub async fn slow(Query(p): Query<SlowParams>) -> &'static str {
    tokio::time::sleep(std::time::Duration::from_secs(p.secs.unwrap_or(30))).await;
    "finally done"
}

pub async fn health_live() -> &'static str {
    "ok"
}

pub async fn health_ready<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
) -> (StatusCode, Json<Value>) {
    let db = repo.ping().await.is_ok();
    let redis_ok = redis::cmd("PING")
        .query_async::<String>(&mut redis.clone())
        .await
        .is_ok();
    let status = if db && redis_ok {
        StatusCode::OK
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    };
    (
        status,
        Json(
            json!({ "database": if db { "up" } else { "down" }, "redis": if redis_ok { "up" } else { "down" } }),
        ),
    )
}

pub async fn metrics_text(Extension(handle): Extension<PrometheusHandle>) -> String {
    handle.render()
}

src/jobs.rs

rust

use crate::auth;
use crate::mail::Mailer;
use crate::repo::Repo;
use tokio::sync::mpsc;
use tokio::time::{Duration, MissedTickBehavior, interval};

pub enum Job {
    Welcome { email: String },
}

pub async fn worker(mut rx: mpsc::Receiver<Job>, mailer: Mailer) {
    while let Some(job) = rx.recv().await {
        match job {
            Job::Welcome { email } => match mailer.welcome(&email).await {
                Ok(()) => tracing::info!(%email, "welcome mail sent"),
                Err(err) => tracing::warn!(?err, %email, "welcome mail failed"),
            },
        }
    }
}

pub async fn cleanup<R: Repo>(repo: R, every: Duration) {
    let mut ticker = interval(every);
    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
    loop {
        ticker.tick().await;
        match repo.delete_expired_refresh(auth::now()).await {
            Ok(n) => tracing::info!(removed = n, "cleanup"),
            Err(err) => tracing::warn!(?err, "cleanup failed"),
        }
    }
}

src/mail.rs

rust

use lettre::message::header::ContentType;
use lettre::{AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor};

use crate::config::Config;

#[derive(Clone)]
pub struct Mailer {
    transport: AsyncSmtpTransport<Tokio1Executor>,
    from: String,
}

impl Mailer {
    pub fn new(cfg: &Config) -> Mailer {
        let transport = AsyncSmtpTransport::<Tokio1Executor>::builder_dangerous(&cfg.smtp_host)
            .port(cfg.smtp_port)
            .build();
        Mailer {
            transport,
            from: cfg.mail_from.clone(),
        }
    }

    pub async fn welcome(&self, to: &str) -> anyhow::Result<()> {
        let message = Message::builder()
            .from(self.from.parse()?)
            .to(to.parse()?)
            .subject("Welcome to the LCE blog")
            .header(ContentType::TEXT_PLAIN)
            .body(format!(
                "Hi {to},\n\nthanks for signing up. Happy writing!\n"
            ))?;
        tokio::time::sleep(std::time::Duration::from_secs(2)).await; // pretend the provider is slow
        self.transport.send(message).await?;
        Ok(())
    }
}

src/main.rs

rust

#[cfg(test)]
mod api_tests;
mod auth;
mod cache;
mod config;
mod error;
mod extract;
mod handlers;
mod jobs;
mod mail;
mod models;
mod repo;
mod storage;

use axum::extract::DefaultBodyLimit;
use axum::extract::{MatchedPath, Request};
use axum::http::{HeaderName, HeaderValue, Method, StatusCode, header};
use axum::middleware::{self, Next};
use axum::response::Response;
use axum::{
    Extension, Router,
    routing::{get, post, put},
};
use jobs::Job;
use mail::Mailer;
use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
use models::Comment;
use redis::aio::{ConnectionManager, ConnectionManagerConfig};
use repo::{Repo, SqliteRepo};
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use storage::{LocalStorage, MemStorage, Storage};
use tokio::sync::broadcast;
use tokio::sync::mpsc;
use tokio::time::Duration;
use tower_governor::{GovernorLayer, governor::GovernorConfigBuilder};
use tower_http::cors::CorsLayer;
use tower_http::set_header::SetResponseHeaderLayer;
use tower_http::timeout::TimeoutLayer;
use tower_http::trace::TraceLayer;
use tracing_subscriber::EnvFilter;
use utoipa::openapi::security::{HttpAuthScheme, HttpBuilder, SecurityScheme};
use utoipa::{Modify, OpenApi};
use utoipa_swagger_ui::SwaggerUi;

pub fn app<R: Repo, S: Storage>(
    repo: R,
    store: S,
    jobs: mpsc::Sender<Job>,
    cache: ConnectionManager,
    live: broadcast::Sender<Comment>,
    handle: PrometheusHandle,
) -> Router {
    let cors = CorsLayer::new()
        .allow_origin(HeaderValue::from_static("http://localhost:5500"))
        .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE])
        .allow_headers([header::AUTHORIZATION, header::CONTENT_TYPE])
        .allow_credentials(true);
    let governor = Arc::new(
        GovernorConfigBuilder::default()
            .per_second(2)
            .burst_size(5)
            .finish()
            .unwrap(),
    );
    let auth_routes = Router::new()
        .route("/signup", post(handlers::signup::<R>))
        .route("/login", post(handlers::login::<R>))
        .route("/auth/refresh", post(handlers::refresh::<R>))
        .layer(GovernorLayer::new(governor));
    let streaming = Router::new()
        .route("/posts/{id}/live", get(handlers::live))
        .route("/posts/{id}/events", get(handlers::events));
    Router::new()
        .nest("/api/v1", posts_v1())
        .nest("/api/v2", posts_v2())
        .merge(posts_v1())
        .route("/posts/{id}/restore", post(handlers::restore_post::<R>))
        .route("/me", get(handlers::me::<R>))
        .route("/admin/users", get(handlers::list_users::<R>))
        .route("/auth/logout", post(handlers::logout::<R>))
        .route(
            "/posts/{id}/images",
            post(handlers::upload_image::<R, S>).layer(DefaultBodyLimit::max(10 * 1024 * 1024)),
        )
        .route("/files/{name}", get(handlers::get_file::<S>))
        .route("/debug/slow", get(handlers::slow))
        .route("/health/live", get(handlers::health_live))
        .route("/health/ready", get(handlers::health_ready::<R>))
        .route("/metrics", get(handlers::metrics_text))
        .merge(auth_routes)
        .layer(TimeoutLayer::with_status_code(
            StatusCode::REQUEST_TIMEOUT,
            Duration::from_secs(10),
        ))
        .merge(streaming)
        .merge(SwaggerUi::new("/swagger-ui").url("/api-docs/openapi.json", ApiDoc::openapi()))
        .route_layer(middleware::from_fn(track))
        .with_state(repo)
        .layer(Extension(Arc::new(store)))
        .layer(Extension(jobs))
        .layer(Extension(cache))
        .layer(Extension(live))
        .layer(Extension(handle))
        .layer(cors)
        .layer(SetResponseHeaderLayer::if_not_present(
            header::X_CONTENT_TYPE_OPTIONS,
            HeaderValue::from_static("nosniff"),
        ))
        .layer(SetResponseHeaderLayer::if_not_present(
            header::X_FRAME_OPTIONS,
            HeaderValue::from_static("DENY"),
        ))
        .layer(SetResponseHeaderLayer::if_not_present(
            header::REFERRER_POLICY,
            HeaderValue::from_static("no-referrer"),
        ))
        .layer(TraceLayer::new_for_http())
}

#[tokio::main]
async fn main() {
    let cfg = config::config();
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
        )
        .init();
    let addr = format!("{}:{}", cfg.host, cfg.port);
    let options = SqliteConnectOptions::from_str(&cfg.database_url)
        .unwrap()
        .create_if_missing(true)
        .journal_mode(SqliteJournalMode::Wal);
    let pool = SqlitePoolOptions::new()
        .max_connections(5)
        .connect_with(options)
        .await
        .unwrap();
    sqlx::migrate!().run(&pool).await.unwrap();
    let repo = SqliteRepo { pool };
    let (tx, rx) = mpsc::channel::<Job>(100);
    let mailer = Mailer::new(cfg);
    tokio::spawn(jobs::worker(rx, mailer));
    tokio::spawn(jobs::cleanup(
        repo.clone(),
        Duration::from_secs(cfg.cleanup_secs),
    ));
    let client = redis::Client::open(cfg.redis_url.as_str()).unwrap();
    let redis_config = ConnectionManagerConfig::new()
        .set_number_of_retries(1)
        .set_connection_timeout(Some(Duration::from_millis(300)))
        .set_response_timeout(Some(Duration::from_millis(300)));
    let cache = ConnectionManager::new_with_config(client, redis_config)
        .await
        .unwrap();
    let (live, _) = broadcast::channel::<Comment>(100);
    let handle = PrometheusBuilder::new()
        .set_buckets(&[0.005, 0.01, 0.05, 0.1, 0.5, 1.0])
        .unwrap()
        .install_recorder()
        .unwrap();
    match cfg.storage.as_str() {
        "memory" => {
            serve(
                app(repo, MemStorage::default(), tx, cache, live, handle),
                &addr,
            )
            .await
        }
        _ => {
            serve(
                app(repo, LocalStorage::new("uploads"), tx, cache, live, handle),
                &addr,
            )
            .await
        }
    }
}

async fn serve(app: Router, addr: &str) {
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    tracing::info!(%addr, "listening");
    axum::serve(
        listener,
        app.into_make_service_with_connect_info::<SocketAddr>(),
    )
    .with_graceful_shutdown(shutdown_signal())
    .await
    .unwrap();
}

fn posts_common<R: Repo>() -> Router<R> {
    Router::new()
        .route(
            "/posts",
            get(handlers::list_posts::<R>).post(handlers::create_post::<R>),
        )
        .route(
            "/posts/{id}",
            put(handlers::update_post::<R>).delete(handlers::delete_post::<R>),
        )
        .route(
            "/posts/{id}/comments",
            get(handlers::list_comments::<R>).post(handlers::add_comment::<R>),
        )
        .route("/search", get(handlers::search_posts::<R>))
}

fn posts_v1<R: Repo>() -> Router<R> {
    posts_common()
        .route("/posts/{id}", get(handlers::get_post::<R>))
        .layer(SetResponseHeaderLayer::overriding(
            HeaderName::from_static("deprecation"),
            HeaderValue::from_static("true"),
        ))
        .layer(SetResponseHeaderLayer::overriding(
            HeaderName::from_static("sunset"),
            HeaderValue::from_static("Wed, 30 Jun 2027 00:00:00 GMT"),
        ))
        .layer(SetResponseHeaderLayer::overriding(
            header::LINK,
            HeaderValue::from_static(r#"</api/v2>; rel="successor-version""#),
        ))
}

fn posts_v2<R: Repo>() -> Router<R> {
    posts_common().route("/posts/{id}", get(handlers::get_post_v2::<R>))
}

async fn shutdown_signal() {
    let ctrl_c = async {
        tokio::signal::ctrl_c().await.expect("Ctrl+C handler");
    };
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("SIGTERM handler")
            .recv()
            .await;
    };
    tokio::select! {
        _ = ctrl_c => {},
        _ = terminate => {},
    }
    tracing::info!("shutting down");
}

async fn track(req: Request, next: Next) -> Response {
    let start = std::time::Instant::now();
    let method = req.method().to_string();
    let path = req
        .extensions()
        .get::<MatchedPath>()
        .map(|p| p.as_str().to_string())
        .unwrap_or_else(|| "unmatched".to_string());
    let response = next.run(req).await;
    let status = response.status().as_u16().to_string();
    let labels = [("method", method), ("path", path), ("status", status)];
    metrics::counter!("http_requests_total", &labels).increment(1);
    metrics::histogram!("http_request_duration_seconds", &labels)
        .record(start.elapsed().as_secs_f64());
    response
}

#[derive(OpenApi)]
#[openapi(
    paths(handlers::get_post_v2, handlers::create_post),
    components(schemas(models::PostV2, models::NewPost, models::PostWithTags, models::Post, error::ErrorBody)),
    modifiers(&SecurityAddon)
)]
struct ApiDoc;

struct SecurityAddon;

impl Modify for SecurityAddon {
    fn modify(&self, openapi: &mut utoipa::openapi::OpenApi) {
        let components = openapi.components.get_or_insert_with(Default::default);
        components.add_security_scheme(
            "bearer",
            SecurityScheme::Http(
                HttpBuilder::new()
                    .scheme(HttpAuthScheme::Bearer)
                    .bearer_format("JWT")
                    .build(),
            ),
        );
    }
}

src/models.rs

rust

use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use validator::{Validate, ValidationError};

#[derive(Serialize, Deserialize, Clone, sqlx::FromRow, ToSchema)]
pub struct Post {
    pub id: i64,
    pub title: String,
    pub body: String,
    pub slug: String,
    pub author_id: Option<i64>,
    pub created_at: String,
}

#[derive(Deserialize, Validate, ToSchema)]
pub struct NewPost {
    #[validate(
        length(min = 1, max = 200, message = "1 to 200 characters"),
        custom(function = "not_blank")
    )]
    pub title: String,
    #[validate(length(min = 1, max = 20000, message = "must not be empty"))]
    pub body: String,
    #[serde(default)]
    #[validate(length(max = 5, message = "at most 5 tags"))]
    pub tags: Vec<String>,
}

#[derive(Serialize, Clone)]
pub struct Comment {
    pub id: i64,
    pub post_id: i64,
    pub body: String,
    pub created_at: String,
}

#[derive(Deserialize, Validate)]
pub struct NewComment {
    #[validate(
        length(min = 1, max = 2000, message = "1 to 2000 characters"),
        custom(function = "not_blank")
    )]
    pub body: String,
}

pub fn slugify(title: &str) -> String {
    let mut slug = String::new();
    for c in title.trim().to_lowercase().chars() {
        if c.is_alphanumeric() {
            slug.push(c);
        } else if !slug.ends_with('-') {
            slug.push('-');
        }
    }
    let slug = slug.trim_end_matches('-').to_string();
    if slug.is_empty() {
        return format!("post-{}", &uuid::Uuid::new_v4().simple().to_string()[..8]);
    }
    slug
}

#[derive(Serialize, ToSchema)]
pub struct PostWithTags {
    #[serde(flatten)]
    pub post: Post,
    pub tags: Vec<String>,
}

#[derive(Deserialize)]
pub struct ListParams {
    pub page: Option<u32>,
    pub per_page: Option<u32>,
    pub tag: Option<String>,
    #[serde(default)]
    pub sort: Sort,
}

#[derive(Deserialize, Clone, Copy, Default)]
#[serde(rename_all = "lowercase")]
pub enum Sort {
    #[default]
    Newest,
    Oldest,
    Title,
}

#[derive(Serialize)]
pub struct Page<T> {
    pub items: Vec<T>,
    pub page: u32,
    pub per_page: u32,
    pub total: i64,
}

pub struct ListFilter {
    pub page: u32,
    pub per_page: u32,
    pub tag: Option<String>,
    pub sort: Sort,
}

#[derive(Serialize)]
pub struct Hit {
    pub id: i64,
    pub title: String,
    pub snippet: String,
}

#[derive(Deserialize)]
pub struct SearchParams {
    pub q: String,
}

pub fn not_blank(value: &str) -> Result<(), ValidationError> {
    if value.trim().is_empty() {
        return Err(ValidationError::new("blank").with_message("must not be blank".into()));
    }
    Ok(())
}

#[derive(Deserialize, Validate)]
pub struct NewUser {
    #[validate(email(message = "must be a valid email"))]
    pub email: String,
    #[validate(length(min = 10, max = 200, message = "10 to 200 characters"))]
    pub password: String,
}

#[derive(Serialize, Clone)]
pub struct User {
    pub id: i64,
    pub email: String,
    pub role: String,
    pub created_at: String,
}

pub struct UserWithHash {
    pub id: i64,
    pub email: String,
    pub password_hash: String,
    pub role: String,
}

#[derive(Serialize)]
pub struct Image {
    pub id: i64,
    pub post_id: i64,
    pub url: String,
}

#[derive(Serialize, ToSchema)]
pub struct PostV2 {
    pub id: i64,
    #[schema(example = "Hello")]
    pub title: String,
    /// The post's text in Markdown.
    pub content: String,
    pub slug: String,
    pub author_id: Option<i64>,
    pub created_at: String,
}

impl From<Post> for PostV2 {
    fn from(p: Post) -> PostV2 {
        PostV2 {
            id: p.id,
            title: p.title,
            content: p.body,
            slug: p.slug,
            author_id: p.author_id,
            created_at: p.created_at,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn slug_from_title() {
        assert_eq!(slugify("Hello, World"), "hello-world");
    }

    #[test]
    fn slug_never_empty() {
        assert!(!slugify("!!!").is_empty());
    }
}

src/repo.rs

rust

use std::future::Future;

use sqlx::{QueryBuilder, Sqlite, SqlitePool};

use crate::models::{
    Comment, Hit, Image, ListFilter, NewComment, NewPost, Post, PostWithTags, Sort, User,
    UserWithHash, slugify,
};

pub trait Repo: Clone + Send + Sync + 'static {
    fn list(&self, f: ListFilter) -> impl Future<Output = sqlx::Result<(Vec<Post>, i64)>> + Send;
    fn get(&self, id: i64) -> impl Future<Output = sqlx::Result<Option<Post>>> + Send;
    fn create(
        &self,
        author_id: i64,
        new: NewPost,
    ) -> impl Future<Output = sqlx::Result<PostWithTags>> + Send;
    fn update(
        &self,
        id: i64,
        new: NewPost,
    ) -> impl Future<Output = sqlx::Result<Option<Post>>> + Send;
    fn delete(&self, id: i64) -> impl Future<Output = sqlx::Result<bool>> + Send;
    fn restore(&self, id: i64) -> impl Future<Output = sqlx::Result<bool>> + Send;
    fn list_comments(
        &self,
        post_id: i64,
    ) -> impl Future<Output = sqlx::Result<Vec<Comment>>> + Send;
    fn add_comment(
        &self,
        post_id: i64,
        new: NewComment,
    ) -> impl Future<Output = sqlx::Result<Comment>> + Send;
    fn search(&self, q: &str) -> impl Future<Output = sqlx::Result<Vec<Hit>>> + Send;
    fn create_user(
        &self,
        email: &str,
        password_hash: &str,
    ) -> impl Future<Output = sqlx::Result<User>> + Send;
    fn find_user_by_email(
        &self,
        email: &str,
    ) -> impl Future<Output = sqlx::Result<Option<UserWithHash>>> + Send;
    fn get_user(&self, id: i64) -> impl Future<Output = sqlx::Result<Option<User>>> + Send;
    fn store_refresh(
        &self,
        hash: &str,
        user_id: i64,
        expires_at: i64,
    ) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn take_refresh(
        &self,
        hash: &str,
        now: i64,
    ) -> impl Future<Output = sqlx::Result<Option<i64>>> + Send;
    fn delete_refresh(&self, hash: &str) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn delete_expired_refresh(&self, now: i64) -> impl Future<Output = sqlx::Result<u64>> + Send;
    fn list_users(&self) -> impl Future<Output = sqlx::Result<Vec<User>>> + Send;
    fn ping(&self) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn add_image(
        &self,
        post_id: i64,
        url: &str,
    ) -> impl Future<Output = sqlx::Result<Image>> + Send;
}

#[derive(Clone)]
pub struct SqliteRepo {
    pub pool: SqlitePool,
}

impl Repo for SqliteRepo {
    async fn list(&self, f: ListFilter) -> sqlx::Result<(Vec<Post>, i64)> {
        let mut q = QueryBuilder::<Sqlite>::new(
            "SELECT p.id, p.title, p.body, p.slug, p.created_at, p.author_id FROM posts p WHERE p.deleted_at IS NULL",
        );
        if let Some(tag) = &f.tag {
            q.push(" AND p.id IN (SELECT pt.post_id FROM post_tags pt JOIN tags t ON t.id = pt.tag_id WHERE t.name = ")
                .push_bind(tag.clone())
                .push(")");
        }
        q.push(match f.sort {
            Sort::Newest => " ORDER BY p.id DESC",
            Sort::Oldest => " ORDER BY p.id ASC",
            Sort::Title => " ORDER BY p.title ASC",
        });
        q.push(" LIMIT ").push_bind(f.per_page as i64);
        q.push(" OFFSET ")
            .push_bind(((f.page - 1) * f.per_page) as i64);
        let posts = q.build_query_as::<Post>().fetch_all(&self.pool).await?;

        let mut c =
            QueryBuilder::<Sqlite>::new("SELECT COUNT(*) FROM posts p WHERE p.deleted_at IS NULL");
        if let Some(tag) = &f.tag {
            c.push(" AND p.id IN (SELECT pt.post_id FROM post_tags pt JOIN tags t ON t.id = pt.tag_id WHERE t.name = ")
                .push_bind(tag.clone())
                .push(")");
        }
        let total: i64 = c.build_query_scalar().fetch_one(&self.pool).await?;
        Ok((posts, total))
    }

    async fn search(&self, q: &str) -> sqlx::Result<Vec<Hit>> {
        let q = fts_query(q);
        sqlx::query_as!(
            Hit,
            r#"SELECT p.id as "id!", p.title, snippet(posts_fts, 1, '[', ']', '...', 8) as "snippet!: String"
               FROM posts_fts JOIN posts p ON p.id = posts_fts.rowid
               WHERE posts_fts MATCH ? AND p.deleted_at IS NULL
               ORDER BY bm25(posts_fts) LIMIT 20"#,
            q
        )
        .fetch_all(&self.pool)
        .await
    }

    async fn get(&self, id: i64) -> sqlx::Result<Option<Post>> {
        sqlx::query_as!(Post, "SELECT id, title, body, slug, created_at, author_id FROM posts WHERE id = ? AND deleted_at IS NULL", id)
            .fetch_optional(&self.pool)
            .await
    }

    async fn create(&self, author_id: i64, new: NewPost) -> sqlx::Result<PostWithTags> {
        let slug = slugify(&new.title);
        let mut tx = self.pool.begin().await?;
        let post = sqlx::query_as!(
            Post,
            r#"INSERT INTO posts (title, body, slug, author_id) VALUES (?, ?, ?, ?) RETURNING id, title, body, slug, created_at, author_id"#,
            new.title,
            new.body,
            slug,
            author_id
        )
        .fetch_one(&mut *tx)
        .await?;
        for tag in &new.tags {
            sqlx::query!(
                "INSERT INTO tags (name) VALUES (?) ON CONFLICT (name) DO NOTHING",
                tag
            )
            .execute(&mut *tx)
            .await?;
            sqlx::query!(
                "INSERT INTO post_tags (post_id, tag_id) SELECT ?, id FROM tags WHERE name = ?",
                post.id,
                tag
            )
            .execute(&mut *tx)
            .await?;
        }
        tx.commit().await?;
        Ok(PostWithTags {
            post,
            tags: new.tags,
        })
    }

    async fn update(&self, id: i64, new: NewPost) -> sqlx::Result<Option<Post>> {
        sqlx::query_as!(
            Post,
            r#"UPDATE posts SET title = ?, body = ? WHERE id = ? AND deleted_at IS NULL RETURNING id, title, body, slug, created_at, author_id"#,
            new.title,
            new.body,
            id
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn delete(&self, id: i64) -> sqlx::Result<bool> {
        let result = sqlx::query!(
            "UPDATE posts SET deleted_at = CURRENT_TIMESTAMP WHERE id = ? AND deleted_at IS NULL",
            id
        )
        .execute(&self.pool)
        .await?;
        Ok(result.rows_affected() > 0)
    }

    async fn restore(&self, id: i64) -> sqlx::Result<bool> {
        let result = sqlx::query!(
            "UPDATE posts SET deleted_at = NULL WHERE id = ? AND deleted_at IS NOT NULL",
            id
        )
        .execute(&self.pool)
        .await?;
        Ok(result.rows_affected() > 0)
    }

    async fn create_user(&self, email: &str, password_hash: &str) -> sqlx::Result<User> {
        sqlx::query_as!(
            User,
            r#"INSERT INTO users (email, password_hash) VALUES (?, ?) RETURNING id, email, role, created_at"#,
            email,
            password_hash
        )
        .fetch_one(&self.pool)
        .await
    }

    async fn find_user_by_email(&self, email: &str) -> sqlx::Result<Option<UserWithHash>> {
        sqlx::query_as!(
            UserWithHash,
            "SELECT id, email, password_hash, role FROM users WHERE email = ?",
            email
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn get_user(&self, id: i64) -> sqlx::Result<Option<User>> {
        sqlx::query_as!(
            User,
            "SELECT id, email, role, created_at FROM users WHERE id = ?",
            id
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn store_refresh(&self, hash: &str, user_id: i64, expires_at: i64) -> sqlx::Result<()> {
        sqlx::query!(
            "INSERT INTO refresh_tokens (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
            hash,
            user_id,
            expires_at
        )
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn take_refresh(&self, hash: &str, now: i64) -> sqlx::Result<Option<i64>> {
        sqlx::query_scalar!(
            r#"DELETE FROM refresh_tokens WHERE token_hash = ? AND expires_at > ? RETURNING user_id as "user_id!""#,
            hash,
            now
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn delete_refresh(&self, hash: &str) -> sqlx::Result<()> {
        sqlx::query!("DELETE FROM refresh_tokens WHERE token_hash = ?", hash)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    async fn delete_expired_refresh(&self, now: i64) -> sqlx::Result<u64> {
        let done = sqlx::query!("DELETE FROM refresh_tokens WHERE expires_at < ?", now)
            .execute(&self.pool)
            .await?;
        Ok(done.rows_affected())
    }

    async fn list_comments(&self, post_id: i64) -> sqlx::Result<Vec<Comment>> {
        sqlx::query_as!(
            Comment,
            "SELECT id, post_id, body, created_at FROM comments WHERE post_id = ? ORDER BY id",
            post_id
        )
        .fetch_all(&self.pool)
        .await
    }

    async fn add_comment(&self, post_id: i64, new: NewComment) -> sqlx::Result<Comment> {
        sqlx::query_as!(
            Comment,
            r#"INSERT INTO comments (post_id, body) VALUES (?, ?) RETURNING id as "id!", post_id, body, created_at"#,
            post_id,
            new.body
        )
        .fetch_one(&self.pool)
        .await
    }

    async fn list_users(&self) -> sqlx::Result<Vec<User>> {
        sqlx::query_as!(
            User,
            "SELECT id, email, role, created_at FROM users ORDER BY id"
        )
        .fetch_all(&self.pool)
        .await
    }

    async fn ping(&self) -> sqlx::Result<()> {
        sqlx::query("SELECT 1").execute(&self.pool).await?;
        Ok(())
    }

    async fn add_image(&self, post_id: i64, url: &str) -> sqlx::Result<Image> {
        sqlx::query_as!(
            Image,
            r#"INSERT INTO images (post_id, url) VALUES (?, ?) RETURNING id as "id!", post_id, url"#,
            post_id,
            url
        )
        .fetch_one(&self.pool)
        .await
    }
}

fn fts_query(input: &str) -> String {
    input
        .split_whitespace()
        .map(|word| format!("\"{}\"", word.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" ")
}

src/storage.rs

rust

use std::collections::HashMap;
use std::future::Future;
use std::io;
use std::path::PathBuf;
use std::sync::Mutex;

pub trait Storage: Send + Sync + 'static {
    fn put(&self, name: &str, data: &[u8]) -> impl Future<Output = io::Result<()>> + Send;
    fn get(&self, name: &str) -> impl Future<Output = io::Result<Vec<u8>>> + Send;
    fn delete(&self, name: &str) -> impl Future<Output = io::Result<()>> + Send;
}

pub struct LocalStorage {
    root: PathBuf,
}

impl LocalStorage {
    pub fn new(root: &str) -> LocalStorage {
        std::fs::create_dir_all(root).unwrap();
        LocalStorage {
            root: PathBuf::from(root),
        }
    }

    fn path(&self, name: &str) -> io::Result<PathBuf> {
        if name.is_empty() || name.starts_with('.') || name.contains('/') || name.contains('\\') {
            return Err(io::Error::new(io::ErrorKind::InvalidInput, "bad file name"));
        }
        Ok(self.root.join(name))
    }
}

impl Storage for LocalStorage {
    async fn put(&self, name: &str, data: &[u8]) -> io::Result<()> {
        tokio::fs::write(self.path(name)?, data).await
    }

    async fn get(&self, name: &str) -> io::Result<Vec<u8>> {
        tokio::fs::read(self.path(name)?).await
    }

    async fn delete(&self, name: &str) -> io::Result<()> {
        tokio::fs::remove_file(self.path(name)?).await
    }
}

#[derive(Default)]
pub struct MemStorage {
    files: Mutex<HashMap<String, Vec<u8>>>,
}

impl Storage for MemStorage {
    async fn put(&self, name: &str, data: &[u8]) -> io::Result<()> {
        self.files
            .lock()
            .unwrap()
            .insert(name.to_string(), data.to_vec());
        Ok(())
    }

    async fn get(&self, name: &str) -> io::Result<Vec<u8>> {
        self.files
            .lock()
            .unwrap()
            .get(name)
            .cloned()
            .ok_or(io::Error::from(io::ErrorKind::NotFound))
    }

    async fn delete(&self, name: &str) -> io::Result<()> {
        self.files.lock().unwrap().remove(name);
        Ok(())
    }
}

Recap

A workflow in .github/workflows/ci.yml runs on every push and pull request
cargo fmt --check, cargo clippy -- -D warnings and cargo test: each found a real problem
Values the tests need go in env:; real secrets go in the repository's secrets
A second job with needs: and if: builds and pushes the image to ghcr.io with GITHUB_TOKEN
Deploy = pull the new image and run it

Next episode: that was Part 5: our blog API is in production. Part 6 goes deeper into Rust itself, starting with macros. Watch the series on YouTube.