9 files changed, 270 insertions, 129 deletions

diff --git a/lib/api-config/Cargo.toml b/lib/api-config/Cargo.toml
index b114944..6724306 100644
--- a/lib/api-config/Cargo.toml
+++ b/lib/api-config/Cargo.toml
@@ -18,6 +18,9 @@ tracing.workspace = true
 utoipa = { workspace = true, optional = true }
 warden-core.workspace = true
 
+[dev-dependencies]
+anyhow.workspace = true
+
 [features]
 default = []
 utoipa = ["dep:utoipa", "utoipa/time"]
diff --git a/lib/api-config/src/lib.rs b/lib/api-config/src/lib.rs
index 84366c1..0d0117a 100644
--- a/lib/api-config/src/lib.rs
+++ b/lib/api-config/src/lib.rs
@@ -1,3 +1,5 @@
 mod error;
 pub mod schema;
 pub use error::ConfigurationError;
+
+pub(crate) static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations");
diff --git a/lib/api-config/src/schema/create.rs b/lib/api-config/src/schema/create_schema.rs
index e6511d5..493fb09 100644
--- a/lib/api-config/src/schema/create.rs
+++ b/lib/api-config/src/schema/create_schema.rs
@@ -1,4 +1,10 @@
 use serde::{Deserialize, Serialize};
+use tracing::debug;
+
+use crate::{
+    ConfigurationError,
+    schema::{SchemaService, TransactionSchema},
+};
 
 #[derive(Deserialize, Serialize)]
 #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
@@ -41,3 +47,24 @@ pub struct CreateSchema {
     /// The json schema
     pub schema: serde_json::Value,
 }
+
+pub(super) async fn create_schema(
+    state: &SchemaService,
+    kind: &str,
+    version: &str,
+    schema: &serde_json::Value,
+) -> Result<TransactionSchema, ConfigurationError> {
+    debug!("creating transaction schema");
+    sqlx::query_as!(
+        TransactionSchema,
+        "insert into transaction_schema (schema_type, schema_version, schema) values ($1, $2, $3)
+            returning *
+            ",
+        kind,
+        version,
+        sqlx::types::Json(&schema) as _
+    )
+    .fetch_one(&state.database)
+    .await
+    .map_err(|e| e.into())
+}
diff --git a/lib/api-config/src/schema/delete_schema.rs b/lib/api-config/src/schema/delete_schema.rs
new file mode 100644
index 0000000..86e71d5
--- /dev/null
+++ b/lib/api-config/src/schema/delete_schema.rs
@@ -0,0 +1,19 @@
+use tracing::debug;
+
+use crate::{ConfigurationError, schema::SchemaService};
+
+pub(super) async fn delete_schema(
+    state: &SchemaService,
+    kind: &str,
+    version: &str,
+) -> Result<(), ConfigurationError> {
+    debug!("deleting transaction schema");
+    sqlx::query!(
+        "delete from transaction_schema where schema_type = $1 and schema_version = $2",
+        kind,
+        version,
+    )
+    .execute(&state.database)
+    .await?;
+    Ok(())
+}
diff --git a/lib/api-config/src/schema/get_schema.rs b/lib/api-config/src/schema/get_schema.rs
new file mode 100644
index 0000000..427a02f
--- /dev/null
+++ b/lib/api-config/src/schema/get_schema.rs
@@ -0,0 +1,26 @@
+use tracing::debug;
+
+use crate::{
+    ConfigurationError,
+    schema::{SchemaService, TransactionSchema},
+};
+
+pub(super) async fn get_schema(
+    state: &SchemaService,
+    kind: &str,
+    version: &str,
+) -> Result<Option<TransactionSchema>, ConfigurationError> {
+    debug!("getting transaction schema");
+    let result = sqlx::query_as!(
+        TransactionSchema,
+        "select 
+            *
+            from transaction_schema where schema_type = $1 and schema_version = $2",
+        kind,
+        version,
+    )
+    .fetch_optional(&state.database)
+    .await?;
+
+    Ok(result)
+}
diff --git a/lib/api-config/src/schema/implementation.rs b/lib/api-config/src/schema/implementation.rs
new file mode 100644
index 0000000..c414879
--- /dev/null
+++ b/lib/api-config/src/schema/implementation.rs
@@ -0,0 +1,84 @@
+use async_trait::async_trait;
+use tracing::debug;
+
+use crate::schema::{self, SchemaDriver, SchemaService, TransactionSchema};
+
+#[async_trait]
+impl SchemaDriver for SchemaService {
+    #[tracing::instrument(skip(self, schema))]
+    async fn create_schema(
+        &self,
+        kind: &str,
+        version: &str,
+        schema: &serde_json::Value,
+    ) -> Result<TransactionSchema, crate::ConfigurationError> {
+        schema::create_schema::create_schema(self, kind, version, schema).await
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn delete_schema(
+        &self,
+        kind: &str,
+        version: &str,
+    ) -> Result<(), crate::ConfigurationError> {
+        schema::delete_schema::delete_schema(self, kind, version).await
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn get_schema(
+        &self,
+        kind: &str,
+        version: &str,
+    ) -> Result<Option<TransactionSchema>, crate::ConfigurationError> {
+        schema::get_schema::get_schema(self, kind, version).await
+    }
+
+    #[tracing::instrument(skip(self, schema))]
+    async fn update_schema(
+        &self,
+        kind: &str,
+        version: &str,
+        schema: &serde_json::Value,
+    ) -> Result<Option<TransactionSchema>, crate::ConfigurationError> {
+        schema::update_schema::update_schema(self, kind, version, schema).await
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn get_schemas(
+        &self,
+        limit: i64,
+        first: Option<i64>,
+        after: Option<&str>,
+    ) -> Result<Vec<TransactionSchema>, crate::ConfigurationError> {
+        debug!("getting transaction schemas");
+        let limit = first.unwrap_or(limit);
+        let mut last_type = String::default();
+        let mut last_version = String::default();
+
+        if let Some(s) = after {
+            let parts: Vec<&str> = s.split(',').collect();
+            if parts.len() == 2 {
+                last_type = parts[0].to_string();
+                last_version = parts[1].to_string();
+            }
+        }
+
+        let rows = sqlx::query_as!(
+            TransactionSchema,
+            "
+            select *
+            from transaction_schema
+            where ($1 = '' or (schema_type, schema_version) > ($1, $2))
+            order by schema_type asc, schema_version asc
+            limit $3
+        ",
+            &last_type,
+            &last_version,
+            limit + 1
+        )
+        .fetch_all(&self.database)
+        .await?;
+
+        Ok(rows)
+    }
+}
diff --git a/lib/api-config/src/schema/mod.rs b/lib/api-config/src/schema/mod.rs
index f626e87..d16ee9f 100644
--- a/lib/api-config/src/schema/mod.rs
+++ b/lib/api-config/src/schema/mod.rs
@@ -1,7 +1,10 @@
-mod create;
-pub use create::CreateSchema;
+mod create_schema;
+mod delete_schema;
+mod get_schema;
+mod implementation;
+mod update_schema;
+pub use create_schema::CreateSchema;
 use sqlx::PgPool;
-use tracing::debug;
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
@@ -96,133 +99,64 @@ pub trait SchemaDriver: Send + Sync {
     ) -> Result<Vec<TransactionSchema>, ConfigurationError>;
 }
 
-#[async_trait]
-impl SchemaDriver for SchemaService {
-    #[tracing::instrument(skip(self, schema))]
-    async fn create_schema(
-        &self,
-        kind: &str,
-        version: &str,
-        schema: &serde_json::Value,
-    ) -> Result<TransactionSchema, crate::ConfigurationError> {
-        debug!("creating transaction schema");
-        sqlx::query_as!(
-            TransactionSchema,
-            "insert into transaction_schema (schema_type, schema_version, schema) values ($1, $2, $3)
-            returning *
-            ",
-            kind,
-            version,
-            sqlx::types::Json(&schema) as _
-        )
-        .fetch_one(&self.database)
-        .await
-        .map_err(|e| e.into())
-    }
+#[cfg(test)]
+mod tests {
+    use sqlx::PgPool;
+
+    use crate::schema::{SchemaDriver, SchemaService};
+
+    #[sqlx::test(
+        migrator = "crate::MIGRATOR",
+        fixtures(path = "../../tests/fixtures", scripts("schema"))
+    )]
+    async fn schema(pool: PgPool) -> anyhow::Result<()> {
+        let driver = SchemaService { database: pool };
+
+        // 2. Define Fixtures
+        let kind = "fin_tx_v1";
+        let version = "1.0.0";
+        let min_schema = serde_json::json!({
+            "type": "object",
+            "required": ["amount", "currency"],
+            "properties": {
+                "amount": { "type": "integer" },
+                "ccy": { "type": "string" }
+            }
+        });
+
+        // CREATE
+        let created = driver
+            .create_schema(kind, version, &min_schema)
+            .await
+            .expect("Create failed");
+        assert_eq!(created.schema_type, kind);
+
+        // GET
+        let found = driver
+            .get_schema("payment", "1.0.0")
+            .await
+            .expect("Get failed")
+            .expect("Schema missing");
+        assert_eq!(found.schema["required"], min_schema["required"]);
+
+        // UPDATE
+        let updated_json = serde_json::json!({"type": "object", "note": "updated"});
+        let updated = driver
+            .update_schema(kind, version, &updated_json)
+            .await
+            .expect("Update failed")
+            .expect("No row to update");
+        assert_eq!(updated.schema["note"], "updated");
+
+        // DELETE
+        driver
+            .delete_schema(kind, version)
+            .await
+            .expect("Delete failed");
+
+        let final_check = driver.get_schema(kind, version).await.unwrap();
+        assert!(final_check.is_none());
 
-    #[tracing::instrument(skip(self))]
-    async fn delete_schema(
-        &self,
-        kind: &str,
-        version: &str,
-    ) -> Result<(), crate::ConfigurationError> {
-        debug!("deleting transaction schema");
-        sqlx::query!(
-            "delete from transaction_schema where schema_type = $1 and schema_version = $2",
-            kind,
-            version,
-        )
-        .execute(&self.database)
-        .await?;
         Ok(())
     }
-
-    #[tracing::instrument(skip(self))]
-    async fn get_schema(
-        &self,
-        kind: &str,
-        version: &str,
-    ) -> Result<Option<TransactionSchema>, crate::ConfigurationError> {
-        debug!("getting transaction schema");
-        let result = sqlx::query_as!(
-            TransactionSchema,
-            "select 
-            *
-            from transaction_schema where schema_type = $1 and schema_version = $2",
-            kind,
-            version,
-        )
-        .fetch_optional(&self.database)
-        .await?;
-
-        Ok(result)
-    }
-
-    #[tracing::instrument(skip(self, schema))]
-    async fn update_schema(
-        &self,
-        kind: &str,
-        version: &str,
-        schema: &serde_json::Value,
-    ) -> Result<Option<TransactionSchema>, crate::ConfigurationError> {
-        debug!("updating transaction schema");
-        sqlx::query_as!(
-            TransactionSchema,
-            "
-            update
-                transaction_schema
-            set 
-                schema = $3
-            where 
-                schema_type = $1 
-                and schema_version = $2
-            returning *
-        ",
-            kind,
-            version,
-            sqlx::types::Json(&schema) as _
-        )
-        .fetch_optional(&self.database)
-        .await
-        .map_err(|e| e.into())
-    }
-
-    #[tracing::instrument(skip(self))]
-    async fn get_schemas(
-        &self,
-        limit: i64,
-        first: Option<i64>,
-        after: Option<&str>,
-    ) -> Result<Vec<TransactionSchema>, ConfigurationError> {
-        debug!("getting transaction schemas");
-        let limit = first.unwrap_or(limit);
-        let mut last_type = String::default();
-        let mut last_version = String::default();
-
-        if let Some(s) = after {
-            let parts: Vec<&str> = s.split(',').collect();
-            if parts.len() == 2 {
-                last_type = parts[0].to_string();
-                last_version = parts[1].to_string();
-            }
-        }
-
-        let rows = sqlx::query_as!(
-            TransactionSchema,
-            "
-            select *
-            from transaction_schema
-            where ($1 = '' or (schema_type, schema_version) > ($1, $2))
-            order by schema_type asc, schema_version asc
-            limit $3
-        ",
-            &last_type,
-            &last_version,
-            limit + 1
-        )
-        .fetch_all(&self.database)
-        .await?;
-
-        Ok(rows)
-    }
 }
diff --git a/lib/api-config/src/schema/update_schema.rs b/lib/api-config/src/schema/update_schema.rs
new file mode 100644
index 0000000..4aa0862
--- /dev/null
+++ b/lib/api-config/src/schema/update_schema.rs
@@ -0,0 +1,34 @@
+use tracing::debug;
+
+use crate::{
+    ConfigurationError,
+    schema::{SchemaService, TransactionSchema},
+};
+
+pub(super) async fn update_schema(
+    state: &SchemaService,
+    kind: &str,
+    version: &str,
+    schema: &serde_json::Value,
+) -> Result<Option<TransactionSchema>, ConfigurationError> {
+    debug!("updating transaction schema");
+    sqlx::query_as!(
+        TransactionSchema,
+        "
+            update
+                transaction_schema
+            set 
+                schema = $3
+            where 
+                schema_type = $1 
+                and schema_version = $2
+            returning *
+        ",
+        kind,
+        version,
+        sqlx::types::Json(&schema) as _
+    )
+    .fetch_optional(&state.database)
+    .await
+    .map_err(|e| e.into())
+}
diff --git a/lib/api-config/tests/fixtures/schema.sql b/lib/api-config/tests/fixtures/schema.sql
new file mode 100644
index 0000000..4ec082e
--- /dev/null
+++ b/lib/api-config/tests/fixtures/schema.sql
@@ -0,0 +1,12 @@
+insert into transaction_schema (schema_type, schema_version, schema)
+values 
+(
+    'payment', 
+    '1.0.0', 
+    '{"type": "object", "required": ["amount", "currency"], "properties": {"amount": {"type": "number"}, "currency": {"type": "string"}}}'
+),
+(
+    'refund', 
+    '1.0.0', 
+    '{"type": "object", "required": ["original_txn_id"], "properties": {"original_txn_id": {"type": "string"}}}'
+);