add rough DB stuff

.env

@@ -0,0 +1 @@
+DATABASE_URL=postgres://ztoa:0zpVXAVK20@localhost:5432/ztoa

.gitignore

@@ -1 +1,2 @@
 /target
+*.db

Cargo.toml

@@ -17,12 +17,18 @@ hyper = "1.1"
 pin-project = "1.1.0"
 serde = { version = "1.0.164", features = ["derive"] }
 serde_json = "1.0.99"
+sqlx = { version = "0.8.5", features = ["runtime-tokio", "macros", "postgres", "uuid", "chrono", "migrate"] }
+tar = "0.4.44"
 thiserror = "2.0"
 tokio = { version = "1.28.2", features = ["full"] }
 tokio-stream = "0.1"
+tokio-util = "0.7.15"
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3", features =["env-filter"] }
+uuid = { version = "1.16.0", features = ["v4"] }
 
 [dev-dependencies]
+bollard = { git = "https://github.com/fussybeaver/bollard.git", rev = "50a25a0" }
+http-body-util = "0.1.3"
 reqwest = { version = "0.12", features = ["cookies"] }
 test-log = { version = "0.2.12", default-features = false, features = ["trace"] }

conf/default.toml

@@ -3,4 +3,4 @@ debug = true
 listen = "[::]:3742"
 
 [database]
-url = "sqlite://./zero-to-axum.db"
+url = "postgres://ztoa:0zpVXAVK20@localhost:5432/ztoa"

migrations/20250514191325_create-subscriptions-table.sql

@@ -0,0 +1,7 @@
+-- Create Subscriptions Table
+CREATE TABLE subscriptions(
+    id uuid NOT NULL PRIMARY KEY,
+    email TEXT NOT NULL UNIQUE,
+    name TEXT,
+    subscribed_at timestamptz NOT NULL
+)

scripts/start_dev_db.sh

@@ -0,0 +1,50 @@
+#!/bin/bash
+
+SUPERUSER_NAME="postgres"
+SUPERUSER_PASS="nOMvDptXFk"
+
+USER_NAME="ztoa"
+USER_PASS="0zpVXAVK20"
+
+APP_DB_NAME="ztoa"
+
+CONTAINER_NAME="postgres"
+
+DATABASE_URL="postgres://ztoa:0zpVXAVK20@localhost:5432/ztoa"
+
+# Launch postgres
+podman run \
+  --replace \
+  --env POSTGRES_USER=${SUPERUSER_NAME} \
+  --env POSTGRES_PASSWORD=${SUPERUSER_PASS} \
+  --health-cmd="pg_isready -U ${SUPERUSER_NAME} || exit 1" \
+  --health-interval=1s \
+  --health-timeout=5s \
+  --health-retries=5 \
+  --publish  5432:5432 \
+  --detach \
+  --name "${CONTAINER_NAME}" \
+  postgres -N 1000
+
+# Wait for Postgres to be ready
+until [ \
+  "$(podman inspect -f "{{.State.Health.Status}}" ${CONTAINER_NAME})" == \
+  "healthy" \
+]; do     
+  >&2 echo "Postgres is still unavailable - sleeping"
+  sleep 1 
+done
+
+>&2 echo "Postgres is up and running"
+
+# Create the application user
+CREATE_QUERY="CREATE USER ${USER_NAME} WITH PASSWORD '${USER_PASS}';"
+podman exec -it "${CONTAINER_NAME}" psql -U "${SUPERUSER_NAME}" -c "${CREATE_QUERY}"
+
+# Grant create db privileges to the app user
+GRANT_QUERY="ALTER USER ${USER_NAME} CREATEDB;"
+podman exec -it "${CONTAINER_NAME}" psql -U "${SUPERUSER_NAME}" -c "${GRANT_QUERY}"
+
+sqlx database create
+
+sqlx migrate run

src/conf.rs

@@ -3,12 +3,12 @@ use config::{Config, Environment, File};
 use serde::Deserialize;
 use std::{env, net::SocketAddr};
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct Database {
     pub url: String,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 #[allow(unused)]
 pub struct Conf {
     pub debug: bool,

src/main.rs

@@ -16,7 +16,7 @@ async fn main() -> Result<()> {
     init_tracing();
 
     let conf = Conf::read()?;
-    let server = ZeroToAxum::serve(conf).await;
+    let server = ZeroToAxum::serve(conf).await?;
 
     server.await.context("run server")
 }

src/server/mod.rs

@@ -4,10 +4,13 @@ use anyhow::Result;
 use axum::extract::FromRef;
 use axum_extra::extract::cookie::Key;
 use pin_project::pin_project;
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
 use std::future::{Future, IntoFuture};
 use std::net::SocketAddr;
 use std::pin::pin;
 use std::pin::Pin;
+use std::sync::Arc;
 use tokio::signal;
 use tracing::info;
 
@@ -36,13 +39,20 @@ impl ZeroToAxum {
         self.bound_addr
     }
 
-    pub async fn serve(conf: Conf) -> ZeroToAxum {
+    pub async fn serve(conf: Conf) -> Result<ZeroToAxum> {
-        let state = AppState {
+        let db = PgPoolOptions::new()
+            .max_connections(5)
+            .connect(&conf.database.url)
+            .await?;
+
+        let app_state = AppState {
+            conf: Arc::new(conf.clone()),
             // TODO: pull from config
             key: Key::generate(),
+            db,
         };
 
-        let app = routes::build().with_state(state);
+        let app = routes::build().with_state(app_state);
 
         let listener = tokio::net::TcpListener::bind(&conf.listen).await.unwrap();
         let bound_addr = listener.local_addr().unwrap();
@@ -50,17 +60,20 @@ impl ZeroToAxum {
 
         info!("server started, listening on {bound_addr:?}");
 
-        ZeroToAxum {
+        Ok(ZeroToAxum {
             server: Box::pin(server.into_future()),
             bound_addr,
-        }
+        })
     }
 }
 
 #[derive(Clone)]
 pub struct AppState {
+    #[allow(unused)]
+    conf: Arc<Conf>,
     // The key used to encrypt cookies.
     key: Key,
+    db: PgPool,
 }
 
 impl FromRef<AppState> for Key {

src/server/routes/subscriptions/mod.rs

@@ -1,6 +1,9 @@
-use axum::{http::StatusCode, response::IntoResponse, routing::post, Form, Router};
+use anyhow::Context;
+use axum::{extract::State, http::StatusCode, response::IntoResponse, routing::post, Form, Router};
 use serde::Deserialize;
+use sqlx::types::chrono::Utc;
 use tracing::info;
+use uuid::Uuid;
 
 use crate::server::AppState;
 
@@ -14,24 +17,46 @@ pub struct SubscribeForm {
     email: String,
 }
 
-pub async fn subscribe(Form(form): Form<SubscribeForm>) -> Result<(), SubscribeError> {
+pub async fn subscribe(
+    State(AppState { db, .. }): State<AppState>,
+    Form(form): Form<SubscribeForm>,
+) -> Result<(), SubscribeError> {
     info!(form.name, form.email, "subscribe attempt");
 
     if form.email.is_empty() {
         return Err(SubscribeError::InvalidEmail);
     }
 
+    sqlx::query!(
+        r#"
+            INSERT INTO subscriptions (id, email, name, subscribed_at)
+            VALUES ($1, $2, $3, $4)
+        "#,
+        Uuid::new_v4(),
+        form.email,
+        form.name,
+        Utc::now()
+    )
+    .execute(&db)
+    .await
+    .context("insert subscription into database")?;
+
     Ok(())
 }
 
+#[derive(thiserror::Error, Debug)]
 pub enum SubscribeError {
+    #[error("Invalid Email Address")]
     InvalidEmail,
+    #[error("Unknown Error: {0}")]
+    Unknown(#[from] anyhow::Error),
 }
 
 impl IntoResponse for SubscribeError {
     fn into_response(self) -> axum::response::Response {
         match self {
             SubscribeError::InvalidEmail => (StatusCode::BAD_REQUEST, "Invalid Email Address"),
+            SubscribeError::Unknown(_e) => (StatusCode::INTERNAL_SERVER_ERROR, "Unknown Error"),
         }
         .into_response()
     }

tests/fixture/mod.rs

@@ -1,33 +1,62 @@
 use anyhow::Result;
-use futures_util::FutureExt;
+use bollard::query_parameters::CreateImageOptionsBuilder;
+use bollard::secret::{
+    ContainerCreateBody, ContainerInspectResponse, ContainerState, CreateImageInfo, Health,
+    HealthConfig, HealthStatusEnum, HostConfig, PortBinding,
+};
+use bollard::Docker;
+use futures_util::{FutureExt, StreamExt as _};
+use sqlx::migrate::MigrateDatabase;
+use sqlx::{Connection, PgConnection};
 use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::OnceCell;
 use tokio::task::JoinHandle;
-use tracing::info;
+use tokio::time::sleep;
+use tracing::{debug, trace};
 use zero_to_axum::{Conf, ZeroToAxum};
 
+static SHARED_DB: OnceCell<Arc<TestDb>> = OnceCell::const_new();
+async fn get_shared_db() -> Arc<TestDb> {
+    SHARED_DB
+        .get_or_init(|| async { Arc::new(TestDb::spawn().await) })
+        .await
+        .clone()
+}
+
 pub struct TestServer {
     server_task_handle: JoinHandle<()>,
     addr: SocketAddr,
+    db: Arc<TestDb>,
 }
 
 impl TestServer {
     pub async fn spawn() -> TestServer {
-        info!("start server");
+        debug!("start test server");
+
+        // TODO: allow per-test DBs in some cases
+        // let db = TestDb::spawn().await;
+        // TODO: share test server between test file, somehow?
+        let db = get_shared_db().await;
+        let url = dbg!(db.get_url());
+
         let server = ZeroToAxum::serve(Conf {
             listen: "[::]:0".parse().unwrap(),
-            database: zero_to_axum::conf::Database {
+            database: zero_to_axum::conf::Database { url },
-                url: "memory:".into(),
-            },
             debug: true,
         })
-        .await;
+        .await
+        .unwrap();
         let addr = server.local_addr();
         let server_task_handle = tokio::spawn(server.map(|res| res.unwrap()));
-        info!("server spawned");
+        debug!(?addr, "test server spawned");
 
         TestServer {
             server_task_handle,
             addr,
+            db,
         }
     }
 
@@ -41,6 +70,305 @@ impl TestServer {
         self.server_task_handle.abort();
         let _ = self.server_task_handle.await;
 
+        self.db.stop().await;
+
         Ok(())
     }
 }
+
+const TEST_DB_IMAGE_NAME: &str = "postgres";
+const TEST_DB_SUPERUSER: &str = "postgres";
+const TEST_DB_SUPERUSER_PASS: &str = "password";
+const TEST_DB_APP_USER: &str = "app";
+const TEST_DB_APP_PASS: &str = "apppass";
+const TEST_DB_APP_NAME: &str = "ztoa";
+
+pub struct TestDb {
+    docker: Docker,
+    // image_id: String,
+    container: bollard::secret::ContainerInspectResponse,
+}
+
+impl TestDb {
+    pub async fn spawn() -> Self {
+        let docker = Docker::connect_with_local_defaults().expect("connect to docker daemon");
+
+        let docker = docker.negotiate_version().await.unwrap();
+
+        let version = docker.version().await.unwrap();
+
+        trace!("version: {version:?}");
+
+        let mut image_id = None;
+
+        // check for image
+        if let Ok(image) = docker.inspect_image(TEST_DB_IMAGE_NAME).await {
+            image_id = Some(image.id.unwrap());
+        }
+
+        // build docker image from docker file
+        // let mut image_id = None;
+        // {
+        //     let filename = "Dockerfile.db";
+        //     let image_options = bollard::query_parameters::BuildImageOptionsBuilder::default()
+        //         .dockerfile(filename)
+        //         .rm(true)
+        //         .build();
+
+        //     let archive_bytes = {
+        //         let mut archive = tar::Builder::new(Vec::new());
+        //         archive.append_path(filename).unwrap();
+        //         archive.into_inner().unwrap()
+        //     };
+
+        //     let mut image_build_stream = docker.build_image(
+        //         image_options,
+        //         None,
+        //         Some(http_body_util::Either::Left(http_body_util::Full::new(
+        //             archive_bytes.into(),
+        //         ))),
+        //     );
+
+        //     while let Some(msg) = image_build_stream.next().await {
+        //         info!("Message: {msg:?}");
+
+        //         if let Ok(BuildInfo {
+        //             aux: Some(ImageId { id: Some(id) }),
+        //             ..
+        //         }) = msg
+        //         {
+        //             trace!("Image ID: {id}");
+        //             image_id = Some(id);
+        //         }
+        //     }
+        // }
+        // let image_id = image_id.expect("get image id for built docker image");
+
+        // pull image
+        if image_id.is_none() {
+            let image_opts = CreateImageOptionsBuilder::new()
+                .from_image(TEST_DB_IMAGE_NAME)
+                .build();
+
+            trace!(?image_opts, "pull image");
+            let mut image_create_stream = docker.create_image(Some(image_opts), None, None);
+
+            while let Some(msg) = image_create_stream.next().await {
+                trace!("Message: {msg:?}");
+
+                if let Ok(CreateImageInfo { id: Some(id), .. }) = msg {
+                    trace!("Image ID: {id}");
+                    image_id = Some(id);
+                }
+            }
+        }
+
+        let image_id = image_id.expect("get image id for built docker image");
+
+        // create and start docker container
+        let container_id;
+        {
+            let container_config = ContainerCreateBody {
+                image: Some(image_id.clone()),
+                exposed_ports: Some([("5432/tcp".to_string(), [].into())].into()),
+                host_config: Some(HostConfig {
+                    port_bindings: Some(
+                        [(
+                            "5432/tcp".to_string(),
+                            Some(vec![PortBinding {
+                                host_ip: Some("127.0.0.1".to_string()),
+                                host_port: None, // auto-assign
+                            }]),
+                        )]
+                        .into(),
+                    ),
+                    ..Default::default()
+                }),
+                env: Some(vec![
+                    format!("POSTGRES_USER={TEST_DB_SUPERUSER}"),
+                    format!("POSTGRES_PASSWORD={TEST_DB_SUPERUSER_PASS}"),
+                ]),
+                healthcheck: Some(HealthConfig {
+                    test: Some(vec!["pg_isready -U postgres || exit 1".to_string()]),
+                    // nano seconds
+                    interval: Some(1 * 1000 * 1000 * 1000),
+                    timeout: Some(5 * 1000 * 1000 * 1000),
+                    retries: Some(5 * 1000 * 1000 * 1000),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            };
+
+            trace!("create container");
+            bollard::secret::ContainerCreateResponse {
+                id: container_id,
+                ..
+            } = docker
+                .create_container(
+                    None::<bollard::query_parameters::CreateContainerOptions>,
+                    container_config,
+                )
+                .await
+                .unwrap();
+
+            trace!("start container");
+            docker
+                .start_container(
+                    &container_id,
+                    None::<bollard::query_parameters::StartContainerOptions>,
+                )
+                .await
+                .unwrap();
+        }
+
+        // wait for container to be started
+        let container = loop {
+            trace!("inspect container");
+            let container = docker
+                .inspect_container(
+                    &container_id,
+                    None::<bollard::query_parameters::InspectContainerOptions>,
+                )
+                .await
+                .unwrap();
+
+            if let ContainerInspectResponse {
+                state:
+                    Some(ContainerState {
+                        health:
+                            Some(Health {
+                                status: Some(status),
+                                ..
+                            }),
+                        ..
+                    }),
+                ..
+            } = &container
+            {
+                trace!("status: {status:?}");
+                if *status == HealthStatusEnum::HEALTHY {
+                    break container;
+                }
+            }
+
+            sleep(Duration::from_secs(2)).await;
+        };
+
+        let db = TestDb {
+            docker,
+            // image_id,
+            container,
+        };
+
+        // setup app db
+        {
+            let mut conn = PgConnection::connect(&db.get_superuser_url())
+                .await
+                .unwrap();
+
+            // create application user
+            // Note: In general, string formtting a query is bad practice, but it's required here.
+            sqlx::query(&format!(
+                "CREATE USER {TEST_DB_APP_USER} WITH PASSWORD '{TEST_DB_APP_PASS}';"
+            ))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+
+            // grant privs to app user
+            // Note: In general, string formtting a query is bad practice, but it's required here.
+            sqlx::query(&format!("ALTER USER {TEST_DB_APP_USER} CREATEDB;"))
+                .execute(&mut conn)
+                .await
+                .unwrap();
+        }
+
+        // create test db
+        sqlx::Postgres::create_database(&dbg!(db.get_url()))
+            .await
+            .unwrap();
+
+        let mut conn = PgConnection::connect(&db.get_url()).await.unwrap();
+
+        // run migrations on test db
+        let m = sqlx::migrate::Migrator::new(Path::new("./migrations"))
+            .await
+            .unwrap();
+        m.run(&mut conn).await.unwrap();
+
+        db
+    }
+
+    /// Get the authenticated URL for accessing the test DB from the host.
+    pub fn get_url(&self) -> String {
+        let binding = self
+            .container
+            .network_settings
+            .as_ref()
+            .unwrap()
+            .ports
+            .as_ref()
+            .unwrap()
+            .get("5432/tcp")
+            .as_ref()
+            .unwrap()
+            .as_ref()
+            .unwrap()
+            .first()
+            .unwrap();
+        let host_ip = binding.host_ip.as_ref().unwrap().clone();
+        let host_port = binding.host_port.as_ref().unwrap().clone();
+        format!("postgres://{TEST_DB_APP_USER}:{TEST_DB_APP_PASS}@{host_ip}:{host_port}/{TEST_DB_APP_NAME}")
+    }
+
+    /// Get the superuser-authenticated URL for accessing the `postgres` db from the host.
+    fn get_superuser_url(&self) -> String {
+        let binding = self
+            .container
+            .network_settings
+            .as_ref()
+            .unwrap()
+            .ports
+            .as_ref()
+            .unwrap()
+            .get("5432/tcp")
+            .as_ref()
+            .unwrap()
+            .as_ref()
+            .unwrap()
+            .first()
+            .unwrap();
+        let host_ip = binding.host_ip.as_ref().unwrap().clone();
+        let host_port = binding.host_port.as_ref().unwrap().clone();
+        format!("postgres://{TEST_DB_SUPERUSER}:{TEST_DB_SUPERUSER_PASS}@{host_ip}:{host_port}/postgres")
+    }
+
+    pub async fn stop(&self) {
+        self.docker
+            .stop_container(
+                &self.container.id.as_deref().unwrap(),
+                #[allow(deprecated)] // is deprecated, but also required
+                None::<bollard::container::StopContainerOptions>,
+            )
+            .await
+            .unwrap();
+
+        self.docker
+            .remove_container(
+                &self.container.id.as_deref().unwrap(),
+                None::<bollard::query_parameters::RemoveContainerOptions>,
+            )
+            .await
+            .unwrap();
+
+        // // TODO: images seem like they might be reused, this is probably a bad idea, but it seems to work?
+        // let _ = self
+        //     .docker
+        //     .remove_image(
+        //         &self.image_id,
+        //         None::<bollard::query_parameters::RemoveImageOptions>,
+        //         None,
+        //     )
+        //     .await;
+    }
+}