I'm currently working on two applications, one a web application and the other a server communicating with various non-browser clients over HTTP, and using Rust for both. I'm using the tide library server-side; I've built a simple service architecture on top allowing me to easily attach services and middleware wherever I need them.

I'm using sqlx to access a Postgres database and the tide-sqlx middleware to provide a database connection to every endpoint. Additional libraries I'll use in the examples below are anyhow, async-std, base64, djangohashers, rand, and serde. I'll also use scopeguard and tide-testing in integration tests.

$ cargo add sqlx tide tide-sqlx anyhow async-std base64 djangohashers rand serde
$ cargo add -D scopeguard tide-testing

My Service trait is incredibly simple, and is only concerned with the attaching of middleware and a Tide server onto another Tide server:

use tide::{self, Middleware};

pub trait Service<State>
    where State: 'static + Clone + Sync + std::marker::Send,
{
    // Or name this `at` to match tide's API.
    fn new(base_route: &str) -> Self;
    fn with_middleware<M: Middleware<State>>(self, middleware: M) -> Self;
    fn register(self, app: &mut tide::Server<State>);
}

An authentication service might look like the code below:

use async_std::prelude::*;
use serde::Deserialize;
use sqlx::{Postgres, Acquire as _};
use tide::{self, Middleware, Request, Response, StatusCode};

use crate::service::Service;


#[derive(Debug)]
pub struct User(pub(crate) i64);

pub trait AuthenticatedRequest {
    // Returns None if the user is not authenticated; otherwise returns their ID.
    fn user_id(&self) -> Option<i64>;
}

impl<State> AuthenticatedRequest for Request<State> {
    fn user_id(&self) -> Option<i64> {
        let u: Option<&User> = self.ext();
        u.map(|v| v.0)
    }
}

pub struct AuthenticationService<State> {
    base: String,
    inner: tide::Server<State>,
}

impl<State> Service<State> for AuthenticationService<State>
    where State: 'static + Default + Clone + Sync + std::marker::Send,
{
    fn new(base_route: &str) -> Self {
        Self {
            base: base_route.to_owned(),
            inner: tide::with_state(State::default()),
        }
    }

    fn with_middleware<M: Sized + Middleware<State>>(mut self, middleware: M)
    -> Self {
        self.inner.with(middleware);
        self
    }

    fn register(mut self, app: &mut tide::Server<State>) {
        app.at(&self.base).nest({
            self.inner.at("/logon").post(logon_user);
            self.inner.at("/logoff").get(logoff_user);
            self.inner
        });
    }
}

#[derive(Debug, Deserialize, Clone)]
struct FormCredential {
    name: String,
    passwd: String,
}

#[derive(sqlx::FromRow, Debug, Deserialize, Clone)]
struct DatabaseCredential {
    id: i64,
    passwd: String,
}

/// Log a user in, creating a new session.
///
/// Sessions are client-specific, so a user may have multiple session tokens at
/// any given time.
///
/// Although a client with a valid token should never call this, doing so is not
/// an error; a new token will be generated, invalidating the current token.
///
/// `logon_user` expects a username and password as form data in the [Request].
///
/// # Warning
///
/// Although we use the Bearer authorization header from OAuth 2.0, this is not 
/// OAuth 2.0. Do not simply copy-paste this code expecting it to be reliable
/// or secure.
async fn logon_user<State>(mut req: Request<State>) -> tide::Result
    where State: 'static + Clone + Send + Sync,
{
    use tide_sqlx::SQLxRequestExt as _;
    use djangohashers::check_password_tolerant;

    let creds: FormCredential = req.body_form().await?;
    let mut conn = req.sqlx_conn::<Postgres>().await;

    let db_creds = sqlx::query_as::<_, DatabaseCredential>(
        "SELECT id, passwd FROM users WHERE name = $1"
    ).bind(&creds.name)
        .fetch_optional(conn.acquire().await?).await
        .map_err(|e|
            tide::Error::new(StatusCode::InternalServerError, e)
        )?;

    if req.header("Authorization").is_some() {
        // TODO: Invalidate current session.
    }

    if let Some(db_creds) = db_creds {
        if check_password_tolerant(&creds.passwd, &db_creds.passwd) {
            let tok = generate_token();
            let conn = conn.acquire();
            let (tok, conn) = tok.join(conn).await;

            let rows = sqlx::query(
                "INSERT INTO sessions (user_id, token) VALUES ($1, $2)"
            ).bind(db_creds.id)
                .bind(&tok)
                .execute(conn?).await
                .map_err(|e|
                    tide::Error::new(StatusCode::InternalServerError, e)
                )?;

            if rows.rows_affected() == 1 {
                Ok(Response::builder(StatusCode::Ok)
                    .body(tok)
                    .build())
            } else {
                Err(tide::Error::from_str(
                    StatusCode::InternalServerError, "User not in database"
                ))
            }
        } else {
            // If credentials don't match, we just send an unauthorized response
            // telling them to log on.
            Ok(Response::builder(StatusCode::Unauthorized)
                .header("WWW-Authenticate", "Bearer")
                .build())
        }
    } else {
        // Incorrect username or unregistered.
        Ok(Response::builder(StatusCode::Unauthorized)
            .header("WWW-Authenticate", "Bearer")
            .build())
    }
}

/// Log a user off, deactivating the current session.
///
/// `logoff_user` expects the client to send the session token in the
/// `Authorization` header.
async fn logoff_user<State>(req: Request<State>) -> tide::Result
    where State: 'static + Clone + Send + Sync,
{
    use tide_sqlx::SQLxRequestExt as _;

    if let Some(token) = req.header("Authorization") {
        let token = token.as_str()
            .split_once(' ')
            .filter(|(h, v)| h == &"Bearer")
            .map(|(h, v)| v);

        let token = if let Some(t) = token {
            t
        } else {
            // We can ignore the lack of an active session.
            return Ok(Response::new(StatusCode::NotFound));
        };

        let mut conn = req.sqlx_conn::<Postgres>().await;

        let rows = sqlx::query(
            "DELETE FROM sessions WHERE token = $1"
        ).bind(token)
            .execute(conn.acquire().await?).await
            .map_err(|e|
                tide::Error::new(StatusCode::InternalServerError, e)
            )?;

        match rows.rows_affected() {
            1 => Ok(Response::new(StatusCode::Ok)),
            _ => Ok(Response::new(StatusCode::NotFound)),
        }
    } else {
        // We can ignore the lack of an active session.
        Ok(Response::new(StatusCode::NotFound))
    }
}

async fn generate_token() -> String {
    use rand::{
        distributions::Alphanumeric,
        Rng, thread_rng,
    };

    let mut rng = thread_rng();

    let s: String = (&mut rng).sample_iter(Alphanumeric)
        .take(20)
        .map(char::from)
        .collect();

    base64::encode(s)
}

The service-related code simply accepts the endpoint at which it will be attached, adds any middleware, then attaches it to another app. We are missing a method -- we need to be able to attach an existing state to the new server, rather than always instantiating a default instance. Since the only state I've used thus far is () I haven't bothered to add this.

The AuthenticationService above passes a token to the client after a successful logon; a session middleware would be responsible for authorization on each request -- it would check for a token in the request header, look it up in the session table (or perhaps an in-memory cache), and either attach the user ID to the request or send an unauthorized response to the client.

A service can be attached to an application like the code below. The database middleware is attached to all endpoints, but the session middleware is only attached to certain services.

mod middleware;
mod service;

use anyhow::anyhow;
use async_std::prelude::*;
use sqlx::postgres::{PgPool, PgPoolOptions};


#[async_std::main]
async fn main() -> anyhow::Result<()> {
    use middleware::ClientSession;
    use service::*;

    let db = SQLxMiddleware::from(init_db_pool().await?);

    let mut app = tide::new();
    app.with(db);

    let sessions = ClientSession::default();

    AuthenticationService::new("/auth/v1")
        .register(&mut app);
    EchoService::new("/echo/v1")
        .with_middleware(sessions)
        .register(&mut app);
    SomeService::new("/some/other/service/v1")
        .with_middleware(sessions)
        .register(&mut app);

    app.listen("127.0.0.1:8080").await?;
    Ok(())
}

async fn init_db_pool() -> PgPool {
    // ...
}

It looks odd that I've versioned each service separately; it's more typical to have a global API version: "/v1/auth", "/v1/echo", etc. I've chosen to version them separately because if I replace a service but version the API globally, then either I need to update all services to v2 even though they haven't changed, or anyone reading my API docs will have to go back and forth between versions (or I duplicate documentation). Versioning each service independently means my documentation can be organized by service rather than API versions.

Integration Testing

I use tide-testing to test my endpoints without the need to spawn a server and make HTTP requests against it. Unfortunately, because of the way sqlx and the sqlx-tide middleware work, I cannot simply create a database transaction, perform my test, and rollback  the transaction. I have to create a new database for each test, then clean it up afterward.

I use an init-db.sql script to create a role and database for manual testing; I'll also use that role for integration tests:

CREATE USER myapp_testing WITH PASSWORD 'myapp_testing' WITH SUPERUSER CREATEDB;
CREATE DATABASE myapp_testing WITH OWNER myapp_testing ENCODING = 'UTF8';

The SUPERUSER privilege will be necessary for our tests; giving SUPERUSER to your production user is a very bad idea.

The above script should be run manually as part of developer and CI setup; I also have a test-data.sql file that I use to import data for manual testing. For the automated test suite, we need to be able to 1) create a database, and 2) remove the database. If postgres is running in a container, step 2 can be skipped. We can add common setup/teardown code to a file in the tests directory:

use sqlx::{
    postgres::{PgConnection, PgPool},
    Connection,
};


fn temp_db_name() -> String {
    use rand::{
        distributions::Alphanumeric,
        Rng,
        thread_rng,
    };

    let rng = thread_rng();

    // The "myapp" name prefix allows us to manually delete all temporary databases
    // with a single query if the automated deletion failed.
    format!("{}_{}", "myapp",
        rng.sample_iter(Alphanumeric)
            .take(7)
            .map(char::from)
            .map(|c| c.to_ascii_lowercase())
            .collect::<String>()
    )
}

/// Create a test database, set up initial data, and return the name of the
/// database and an open connection to it.
pub async fn setup_db() -> anyhow::Result<(String, PgPool)> {
    // In my actual tests, I use OnceCell to store a configuration object from
    // which I get my connection strings.

    // Connection string without a database name, so we can create a database.
    let conn_string = "postgres:://myapp_testing:myapp_testing@localhost";
    let db_name = temp_db_name();

    let mut conn = PgConnection::connect(conn_string).await?;

    sqlx::query(&format!(
        "CREATE DATABASE {} OWNER myapp_testing ENCODING = 'UTF8'", db_name
    ))
    .execute(&mut conn).await?;

    let pool = PgPool::connect(&format!("{}/{}", conn_string, db_name)).await?;
    sqlx::migrate!("./migrations").run(&pool).await?;

    Ok((db_name.to_owned(), pool))
}

pub async fn cleanup_db(name: &str) -> anyhow::Result<()> {
    let conn_string = "postgres:://myapp_testing:myapp_testing@localhost";
    let mut conn = PgConnection::connect(conn_string).await?;
    // Postgresql 13 gives us:
    //sqlx::query(&format!("DROP DATABASE IF EXISTS {} WITH (FORCE)", name))

    // Force-disconnect all clients. This is why we need SUPERUSER.
    sqlx::query("SELECT pg_terminate_backend(pid) FROM pg_stat_activity \
        WHERE datname = $1"
    ).bind(name)
        .execute(&mut conn).await?;

    sqlx::query(&format!("DROP DATABASE IF EXISTS {}", name))
        .execute(&mut conn).await?;

    Ok(())
}

There are two ways to use these; one will always (attempt to) clean up the test database, and the other will only clean up after passing tests, in case you want to be able to inspect the database state after a test fails. In the latter case, add a println!("Creating database {}", db_name); call in your setup_db() function so it will print the database name to your console on test failure.

mod utility;
use utility::{setup_db, cleanup_db};

use my_app::service::{Service, AuthenticationService};

use async_std::task;
use sqlx::postgres::PgPool;
use tide::StatusCode;
use tide_sqlx::SQLxMiddleware;
use tide_testing::{surf, TideTestingExt};

#[async_std::test]
// Always remove the test database.
async fn test_logon_user() -> anyhow::Result<()> {
    let (db, pool) = setup_db().await?;
    let mut _cleanup_db = scopeguard::guard(db, |db| {
        task_block_on(cleanup_db(&db)).unwrap();
    });

    sqlx::query("INSERT INTO users (name, email, passwd) VALUES (\             
        'Favorite Person', 'favorite@example.com', \                             
    'argon2$argon2i$v=19$m=102400,t=2,p=8$K9rz0E1hdrpS$pOulDl1I5kSwWfgOd4h1Vg' \ 
    )").execute(&pool).await?; 


    let mut app = tide::new();
    app.with(SQLxMiddleware::from(pool.clone()));

    AuthenticationService::new("/auth/v1")
        .register(&mut app);

    let mut resp = surf::Response = app.post("/auth/v1/logon")
        .body("name=Favorite%20Person&passwd=favorite").await.unwrap();

    assert_eq!(resp.status(), StatusCode::Ok);

    let body = base64::decode(resp.body_string().await.unwrap())?;               
    let decoded = std::str::from_utf8(&body)?;                                   
                                                                                 
    for c in decoded.chars() {                                                   
        assert!(c.is_alphanumeric());                                            
    }                                                                            
                                                                                 
    Ok(())
}

#[async_std::test]
// Only remove database if the test was successful.
async fn test_logoff_user() -> anyhow::Result<()> {
    let (db, pool) = setup_db().await?;

    sqlx::query("INSERT INTO users (name, email, passwd) VALUES (\
        'Favorite Person', 'favorite@example.com', \
    'argon2$argon2i$v=19$m=102400,t=2,p=8$K9rz0E1hdrpS$pOulDl1I5kSwWfgOd4h1Vg' \
    )").execute(&pool).await?;

    sqlx::query("INSERT INTO sessions (user_id, token) VALUES \
        (1, 'test_token') ON CONFLICT (token) DO NOTHING"
    ).execute(&pool).await?;


    let mut app = tide::new();
    app.with(SQLxMiddleware::from(pool.clone()));

    AuthenticationService::new("/auth/v1")
        .register(&mut app);

    let resp = app.get("/auth/v1/logoff")
        .header("Authorization", "Bearer test_token")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), StatusCode::Ok);

    cleanup_db(&db)?;
    Ok(())
}

This gives us a decent method of integration testing -- the code is simple and clear, and every test is independent of the others. You could even create a tablespace on a ramdisk then create your databases in it to avoid your tests hitting the filesystem.