feat(schema): get schema

11 files changed, 227 insertions, 28 deletions

diff --git a/.sqlx/query-c604b84d88e1a91893127aa66ee5dc15a049639bbabf704fa9e28da0bc66ad07.json b/.sqlx/query-6474a8297762d8df4bc7d09d837cfa2daa0110377d70386b7788d6cd20052a90.json
index a5afb5f..a46ff8f 100644
--- a/.sqlx/query-c604b84d88e1a91893127aa66ee5dc15a049639bbabf704fa9e28da0bc66ad07.json
+++ b/.sqlx/query-6474a8297762d8df4bc7d09d837cfa2daa0110377d70386b7788d6cd20052a90.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "insert into transaction_schema (type, version, json_schema) values ($1, $2, $3)\n            returning\n            type as kind, \n            version, \n            json_schema as schema, \n            created_at, \n            updated_at\n        ",
+  "query": "insert into transaction_schema (type, version, json_schema) values ($1, $2, $3)\n            returning\n                type as kind, \n                version, \n                json_schema as schema, \n                created_at, \n                updated_at\n        ",
   "describe": {
     "columns": [
       {
@@ -44,5 +44,5 @@
       false
     ]
   },
-  "hash": "c604b84d88e1a91893127aa66ee5dc15a049639bbabf704fa9e28da0bc66ad07"
+  "hash": "6474a8297762d8df4bc7d09d837cfa2daa0110377d70386b7788d6cd20052a90"
 }
diff --git a/.sqlx/query-e9e130d0192838e5321e0bbd42a3ffa93d3dd52e4d8736551d328d6a36b9f482.json b/.sqlx/query-e9e130d0192838e5321e0bbd42a3ffa93d3dd52e4d8736551d328d6a36b9f482.json
new file mode 100644
index 0000000..7a35bab
--- /dev/null
+++ b/.sqlx/query-e9e130d0192838e5321e0bbd42a3ffa93d3dd52e4d8736551d328d6a36b9f482.json
@@ -0,0 +1,47 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "select \n        type as kind, \n        version, \n        json_schema as schema, \n        created_at, \n        updated_at\n        from transaction_schema where type = $1 and version = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "kind",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "version",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "schema",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 3,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 4,
+        "name": "updated_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "e9e130d0192838e5321e0bbd42a3ffa93d3dd52e4d8736551d328d6a36b9f482"
+}
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,
diff --git a/warden/src/server/routes/config/mod.rs b/warden/src/server/routes/config/mod.rs
index 0c02eaa..d5344f9 100644
--- a/warden/src/server/routes/config/mod.rs
+++ b/warden/src/server/routes/config/mod.rs
@@ -13,9 +13,10 @@ const CONFIG: &str = "Configuration";
 pub struct ConfigDoc;
 
 pub fn router(store: Arc<AppState>) -> OpenApiRouter {
+    let schema_router = schema::router(store.clone());
+
     OpenApiRouter::new()
         .routes(utoipa_axum::routes!(logs::reload))
-        .routes(utoipa_axum::routes!(schema::create::create_schema))
-        .routes(utoipa_axum::routes!(schema::delete::delete_schema))
         .with_state(store)
+        .merge(schema_router)
 }
diff --git a/warden/src/server/routes/config/schema/create.rs b/warden/src/server/routes/config/schema/create.rs
index 1792df6..9bb3b22 100644
--- a/warden/src/server/routes/config/schema/create.rs
+++ b/warden/src/server/routes/config/schema/create.rs
@@ -10,13 +10,7 @@ use axum::{
 use tracing::{debug, info, trace};
 use warden_core::state::AppState;
 
-use crate::server::{
-    api::{
-        version::Version,
-    },
-    error::AppError,
-    routes::config::CONFIG,
-};
+use crate::server::{api::version::Version, error::AppError, routes::config::CONFIG};
 
 #[utoipa::path(
     put,
diff --git a/warden/src/server/routes/config/schema/delete.rs b/warden/src/server/routes/config/schema/delete.rs
index 51a04da..188f796 100644
--- a/warden/src/server/routes/config/schema/delete.rs
+++ b/warden/src/server/routes/config/schema/delete.rs
@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use api_config::schema::{CreateSchema, SchemaDriver, TransactionSchema};
+use api_config::schema::{CreateSchema, SchemaDriver};
 use axum::{
     Json, debug_handler,
     extract::State,
@@ -10,13 +10,7 @@ use axum::{
 use tracing::debug;
 use warden_core::state::AppState;
 
-use crate::server::{
-    api::{
-        version::Version,
-    },
-    error::AppError,
-    routes::config::CONFIG,
-};
+use crate::server::{api::version::Version, error::AppError, routes::config::CONFIG};
 
 #[utoipa::path(
     delete,
@@ -26,8 +20,7 @@ use crate::server::{
             description = "The schema has been deleted",
             headers(
                 ("x-request-id", description = "Request identifier")
-            ),
-            body = TransactionSchema
+            )
         ),
         (
             status = 400,
diff --git a/warden/src/server/routes/config/schema/mod.rs b/warden/src/server/routes/config/schema/mod.rs
index ab7fa43..1b4fcef 100644
--- a/warden/src/server/routes/config/schema/mod.rs
+++ b/warden/src/server/routes/config/schema/mod.rs
@@ -1,4 +1,17 @@
+use std::sync::Arc;
+
+use utoipa_axum::router::OpenApiRouter;
+use warden_core::state::AppState;
+
 pub mod create;
 pub mod delete;
 pub mod read;
 pub mod update;
+
+pub fn router(store: Arc<AppState>) -> OpenApiRouter {
+    OpenApiRouter::new()
+        .routes(utoipa_axum::routes!(create::create_schema))
+        .routes(utoipa_axum::routes!(delete::delete_schema))
+        .routes(utoipa_axum::routes!(read::get_schema))
+        .with_state(store)
+}
diff --git a/warden/src/server/routes/config/schema/read.rs b/warden/src/server/routes/config/schema/read.rs
index 8b13789..d4e09a7 100644
--- a/warden/src/server/routes/config/schema/read.rs
+++ b/warden/src/server/routes/config/schema/read.rs
@@ -1 +1,116 @@
+use std::sync::Arc;
 
+use api_config::schema::{SchemaDriver, TransactionSchema};
+use axum::{
+    Json, debug_handler,
+    extract::{Query, State},
+    http::{HeaderMap, StatusCode},
+    response::IntoResponse,
+};
+use serde::{Deserialize, Serialize};
+use tracing::debug;
+use utoipa::{IntoParams, ToSchema};
+use warden_core::state::AppState;
+
+use crate::server::{api::version::Version, error::AppError, routes::config::CONFIG};
+
+/// Schema search query
+#[derive(Deserialize, Serialize, IntoParams, ToSchema)]
+#[into_params(parameter_in = Query)]
+pub struct SchemaSearchQuery {
+    /// Schema type
+    #[serde(rename = "type")]
+    kind: String,
+    /// Schema version
+    version: String,
+}
+
+#[utoipa::path(
+    get,
+    responses(
+        (
+            status = 200,
+            description = "Lookup results",
+            headers(
+                ("x-request-id", description = "Request identifier")
+            ),
+            body = Option<TransactionSchema>
+        ),
+        (
+            status = 400,
+            description = "Invalid request",
+            headers(
+                ("x-request-id", description = "Request identifier")
+            )
+        ),
+        (
+            status = 404,
+            description = "Schema not found",
+            headers(
+                ("x-request-id", description = "Request identifier")
+            )
+        ),
+        (
+            status = 405,
+            description = "Method not allowed",
+            headers(
+                ("x-request-id", description = "Request identifier")
+            )
+        ),
+        (
+            status = 500,
+            description = "Internal server error",
+            headers(
+                ("x-request-id", description = "Request identifier")
+            )
+        )
+    ),
+    operation_id = "get_schema", // https://github.com/juhaku/utoipa/issues/1170
+    tag = CONFIG,
+    path = "/{version}/config/schema",
+    params(
+        ("version" = Version, Path, description = "API version, e.g., v1, v2, v3"),
+        SchemaSearchQuery
+    ),
+)]
+#[tracing::instrument(
+    name = "get_schema",
+    skip(state, headers, body),
+    fields(
+        request_id = %headers.get("x-request-id")
+            .and_then(|v| v.to_str().ok()).expect("request id"),
+        kind = %body.kind,
+    )
+)]
+#[debug_handler]
+pub async fn get_schema(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    body: Query<SchemaSearchQuery>,
+) -> Result<impl IntoResponse, AppError> {
+    debug!("searching for schema");
+    // TODO: get from cache
+    let result = state
+        .get_schema(&body.kind, &body.version)
+        .await
+        .map_err(|e| match e {
+            api_config::ConfigurationError::Database(ref error) => match error {
+                sqlx::Error::Database(db_err) if db_err.code() == Some("23505".into()) => {
+                    AppError::new(
+                        StatusCode::CONFLICT,
+                        anyhow::anyhow!("Transaction schema already exists"),
+                    )
+                }
+                _ => e.into(),
+            },
+            _ => e.into(),
+        })?;
+    if let Some(result) = result {
+        Ok(Json(result))
+    } else {
+        Err(AppError::new(
+            StatusCode::NOT_FOUND,
+            anyhow::anyhow!("Schema not found"),
+        ))
+    }
+}