4 files changed, 43 insertions, 7 deletions
diff --git a/lib/api-config/src/schema/mod.rs b/lib/api-config/src/schema/mod.rs
index 4e68129..33d7922 100644
--- a/lib/api-config/src/schema/mod.rs
+++ b/lib/api-config/src/schema/mod.rs
@@ -40,6 +40,12 @@ pub trait SchemaDriver {
         kind: impl AsRef<str> + Send + Sync,
         version: impl AsRef<str> + Send + Sync,
     ) -> Result<(), ConfigurationError>;
+
+    async fn get_schema(
+        &self,
+        kind: impl AsRef<str> + Send + Sync,
+        version: impl AsRef<str> + Send + Sync,
+    ) -> Result<Option<TransactionSchema>, ConfigurationError>;
 }
 
 #[async_trait]
@@ -54,11 +60,11 @@ impl SchemaDriver for AppState {
             TransactionSchema,
             "insert into transaction_schema (type, version, json_schema) values ($1, $2, $3)
             returning
-            type as kind, 
-            version, 
-            json_schema as schema, 
-            created_at, 
-            updated_at
+                type as kind, 
+                version, 
+                json_schema as schema, 
+                created_at, 
+                updated_at
         ",
             kind.as_ref(),
             version.as_ref(),
@@ -74,7 +80,8 @@ impl SchemaDriver for AppState {
         kind: impl AsRef<str> + Send + Sync,
         version: impl AsRef<str> + Send + Sync,
     ) -> Result<(), crate::ConfigurationError> {
-        sqlx::query!("delete from transaction_schema where type = $1 and version = $2",
+        sqlx::query!(
+            "delete from transaction_schema where type = $1 and version = $2",
             kind.as_ref(),
             version.as_ref(),
         )
@@ -82,4 +89,27 @@ impl SchemaDriver for AppState {
         .await?;
         Ok(())
     }
+
+    async fn get_schema(
+        &self,
+        kind: impl AsRef<str> + Send + Sync,
+        version: impl AsRef<str> + Send + Sync,
+    ) -> Result<Option<TransactionSchema>, crate::ConfigurationError> {
+        let result = sqlx::query_as!(
+            TransactionSchema,
+            "select 
+        type as kind, 
+        version, 
+        json_schema as schema, 
+        created_at, 
+        updated_at
+        from transaction_schema where type = $1 and version = $2",
+            kind.as_ref(),
+            version.as_ref(),
+        )
+        .fetch_optional(&self.database)
+        .await?;
+
+        Ok(result)
+    }
 }
diff --git a/lib/warden-core/src/error.rs b/lib/warden-core/src/error.rs
index f05971f..d90a862 100644
--- a/lib/warden-core/src/error.rs
+++ b/lib/warden-core/src/error.rs
@@ -5,6 +5,8 @@ pub enum WardenError {
     #[error(transparent)]
     Datastore(#[from] sqlx::Error),
     #[error(transparent)]
+    Migration(#[from] sqlx::migrate::MigrateError),
+    #[error(transparent)]
     Url(#[from] url::ParseError),
     #[error("Missing required configuration values:\n`{0}`")]
     Config(String),
diff --git a/lib/warden-core/src/state/database.rs b/lib/warden-core/src/state/database.rs
index cf34484..4167424 100644
--- a/lib/warden-core/src/state/database.rs
+++ b/lib/warden-core/src/state/database.rs
@@ -3,7 +3,7 @@ use tracing::{debug, error};
 
 use crate::{WardenError, config::cli::database::Database};
 
-pub async fn connect(config: &Database) -> Result<PgPool, WardenError> {
+pub(crate) async fn connect(config: &Database) -> Result<PgPool, WardenError> {
     let url = config.get_url()?;
     let host = url.host_str();
     debug!(host = host, "connecting to database");
diff --git a/lib/warden-core/src/state/mod.rs b/lib/warden-core/src/state/mod.rs
index f4692c2..18e44b8 100644
--- a/lib/warden-core/src/state/mod.rs
+++ b/lib/warden-core/src/state/mod.rs
@@ -1,5 +1,6 @@
 pub(crate) mod database;
 use sqlx::PgPool;
+use tracing::{debug, trace};
 use tracing_subscriber::EnvFilter;
 
 use crate::{WardenError, config::Configuration};
@@ -15,6 +16,9 @@ pub struct AppState {
 impl AppState {
     pub async fn new(log_handle: LogHandle, config: &Configuration) -> Result<Self, WardenError> {
         let database = database::connect(&config.database).await?;
+        trace!("running database migrations");
+        sqlx::migrate!("../../migrations").run(&database).await?;
+        debug!("database up to date");
 
         Ok(Self {
             log_handle,