1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
|
mod create;
pub use create::CreateSchema;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use warden_core::state::AppState;
use crate::ConfigurationError;
/// Transaction to monitor
#[derive(Deserialize, Serialize)]
#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
pub struct TransactionSchema {
#[serde(rename = "type")]
/// Transaction schema type
pub kind: String,
/// The schema's version
pub version: String,
/// JSON schema for transcation
#[serde(rename = "json_schema")]
pub schema: serde_json::Value,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339")]
pub updated_at: OffsetDateTime,
}
#[async_trait]
pub trait SchemaDriver {
async fn create_schema(
&self,
kind: impl AsRef<str> + Send + Sync,
version: impl AsRef<str> + Send + Sync,
schema: &serde_json::Value,
) -> Result<TransactionSchema, ConfigurationError>;
async fn delete_schema(
&self,
kind: impl AsRef<str> + Send + Sync,
version: impl AsRef<str> + Send + Sync,
) -> Result<(), ConfigurationError>;
}
#[async_trait]
impl SchemaDriver for AppState {
async fn create_schema(
&self,
kind: impl AsRef<str> + Send + Sync,
version: impl AsRef<str> + Send + Sync,
schema: &serde_json::Value,
) -> Result<TransactionSchema, crate::ConfigurationError> {
sqlx::query_as!(
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
",
kind.as_ref(),
version.as_ref(),
sqlx::types::Json(&schema) as _
)
.fetch_one(&self.database)
.await
.map_err(|e| e.into())
}
async fn delete_schema(
&self,
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",
kind.as_ref(),
version.as_ref(),
)
.execute(&self.database)
.await?;
Ok(())
}
}
|
