use axum::body::Body;
use axum::http::{Request, StatusCode};
use chrono::{SubsecRound, Utc};
use http_body_util::BodyExt;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use skilj::{requires_role, CommandType, EventType, Skilj};
use skilj_core::access_control::{self, AccessLevel, Role, RoleAccessMapping, RoleStatus};
use skilj_core::bootstrap::ContextCreator;
use skilj_core::db::{self, Pool};
use skilj_core::event_store::{BoundedContext, BoundedContextStatus, Event};
use skilj_core::plugin::BoundedContextEvent;
use skilj_core::shared::{
generate_token_id, generate_token_secret, CommandDecision, EventSpec, TagMapping,
};
use tower::ServiceExt;
#[derive(Debug, Serialize, Deserialize, JsonSchema)]
struct MoneyDepositedPayload {
amount: i64,
}
struct MoneyDeposited;
impl EventType for MoneyDeposited {
type Payload = MoneyDepositedPayload;
const NAME: &'static str = "MoneyDeposited";
fn direct_creation_allowed() -> bool {
true
}
}
enum BankingEvent {
#[allow(dead_code)]
MoneyDeposited(MoneyDepositedPayload),
}
impl BoundedContextEvent for BankingEvent {
fn try_from_event(event: &Event) -> Option<Result<Self, serde_json::Error>> {
match event.event_type.name.as_str() {
"MoneyDeposited" => {
Some(serde_json::from_str(&event.payload).map(BankingEvent::MoneyDeposited))
}
_ => None,
}
}
}
#[derive(Debug, Serialize, Deserialize, JsonSchema)]
struct WithdrawPayload {
amount: i64,
}
struct WithdrawMoney;
impl CommandType for WithdrawMoney {
type Payload = WithdrawPayload;
type Event = BankingEvent;
const NAME: &'static str = "WithdrawMoney";
fn rest_trigger_allowed() -> bool {
true
}
fn tag_mappings() -> Vec<TagMapping> {
vec![TagMapping {
key: "amount".into(),
field: "amount".into(),
}]
}
fn owner_tag_key() -> Option<&'static str> {
Some("amount")
}
fn decide(payload: &Self::Payload, _matching_events: &[Self::Event]) -> CommandDecision {
if payload.amount > 1000 {
CommandDecision::Rejected {
reason: "insufficient funds".to_string(),
kind: "insufficient_funds".to_string(),
}
} else {
CommandDecision::Accepted {
events: vec![EventSpec {
event_type: "MoneyDeposited".to_string(),
payload: serde_json::json!({ "amount": payload.amount }),
}],
}
}
}
}
struct CloseAccount;
#[requires_role("treasury_officer")]
impl CommandType for CloseAccount {
type Payload = WithdrawPayload;
type Event = BankingEvent;
const NAME: &'static str = "CloseAccount";
fn decide(_payload: &Self::Payload, _matching_events: &[Self::Event]) -> CommandDecision {
CommandDecision::Accepted { events: vec![] }
}
}
struct TestDb {
database_url: String,
pool: Pool,
_embedded: Option<postgresql_embedded::PostgreSQL>,
}
static TEST_DB: tokio::sync::OnceCell<Option<TestDb>> = tokio::sync::OnceCell::const_new();
fn runtime() -> &'static tokio::runtime::Runtime {
static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
RUNTIME.get_or_init(|| {
tokio::runtime::Runtime::new()
.expect("failed to build a tokio runtime for command_trigger tests")
})
}
async fn test_db() -> Option<(String, Pool)> {
TEST_DB
.get_or_init(provision)
.await
.as_ref()
.map(|db| (db.database_url.clone(), db.pool.clone()))
}
async fn connect_and_migrate(database_url: &str, label: &str) -> Option<Pool> {
let pool = match db::connect(database_url).await {
Ok(pool) => pool,
Err(e) => {
eprintln!("skipping: connecting to {label} failed: {e}");
return None;
}
};
if let Err(e) = db::migrate(&pool).await {
eprintln!("skipping: migrating {label} failed: {e}");
return None;
}
Some(pool)
}
async fn provision() -> Option<TestDb> {
let database_url = if let Ok(database_url) = std::env::var("DATABASE_URL") {
database_url
} else {
let mut server = postgresql_embedded::PostgreSQL::default();
if let Err(e) = server.setup().await {
eprintln!(
"skipping: DATABASE_URL not set and embedded PostgreSQL setup failed \
(no network egress to fetch the binary, or a missing system library \
like libxml2 it links against): {e}"
);
return None;
}
if let Err(e) = server.start().await {
eprintln!("skipping: embedded PostgreSQL failed to start: {e}");
return None;
}
let database_name = "skilj_command_trigger_test";
if let Err(e) = server.create_database(database_name).await {
eprintln!("skipping: embedded PostgreSQL create_database failed: {e}");
return None;
}
let url = server.settings().url(database_name);
let pool = connect_and_migrate(&url, "embedded PostgreSQL").await?;
return Some(TestDb {
database_url: url,
pool,
_embedded: Some(server),
});
};
let pool = connect_and_migrate(&database_url, "DATABASE_URL").await?;
Some(TestDb {
database_url,
pool,
_embedded: None,
})
}
fn unique_name(prefix: &str) -> String {
format!("{prefix}_{}", generate_token_id())
}
fn test_now() -> chrono::DateTime<Utc> {
Utc::now().trunc_subsecs(6)
}
async fn setup() -> (
Skilj,
String,
Pool,
String,
RoleAccessMapping,
skilj_core::event_store::CommandType,
) {
let (database_url, pool) = test_db()
.await
.expect("test_db() must be Some - caller already checked");
let external_subject = unique_name("subject");
let role = Role {
id: generate_token_id(),
external_subject: external_subject.clone(),
name: "Reconciliation Role".to_string(),
superadmin: false,
status: RoleStatus::Active,
created_at: test_now(),
revoked_at: None,
};
db::insert_role(&pool, &role).await.unwrap();
let bc_name = unique_name("banking");
let bc = BoundedContext {
name: bc_name.clone(),
status: BoundedContextStatus::Active,
created_at: test_now(),
created_by: ContextCreator::SystemCreator,
template: None,
};
db::insert_bounded_context(&pool, &bc).await.unwrap();
let mapping = RoleAccessMapping {
role: role.clone(),
bounded_context: bc.clone(),
level: AccessLevel::Admin,
can_read_sensitive: false,
scope: None,
status: RoleStatus::Active,
created_at: test_now(),
revoked_at: None,
};
db::insert_role_access_mapping(&pool, &mapping)
.await
.unwrap();
let (skilj, report) = Skilj::builder(database_url)
.pool_options(skilj_core::db::PgPoolOptions::new().max_connections(4))
.bounded_context(bc_name.clone())
.event_type::<MoneyDeposited>()
.command_type::<WithdrawMoney>()
.command_type::<CloseAccount>()
.reconciliation_role(external_subject)
.build()
.await
.unwrap();
assert_eq!(report.skipped_no_access, Vec::<String>::new());
let command_type = db::get_command_type(&pool, &bc_name, "WithdrawMoney")
.await
.unwrap()
.unwrap();
let token = access_control::create_command_token(
&mapping,
&command_type,
generate_token_id(),
generate_token_secret(),
None,
test_now(),
)
.unwrap();
db::insert_command_token(&pool, &token).await.unwrap();
let credential = format!("{}.{}", token.id, token.secret);
(skilj, credential, pool, bc_name, mapping, command_type)
}
#[test]
fn command_trigger_accepts_and_persists_triggered_events() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, credential, pool, bc_name, _, _) = setup().await;
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {credential}"))
.header("content-type", "application/json")
.body(Body::from(r#"{"payload":{"amount":20}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = response.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["accepted"], true);
let sequences = json["triggeredEventSequences"]
.as_array()
.expect("triggeredEventSequences must be present and an array when accepted");
assert_eq!(sequences.len(), 1);
let events = db::list_events_for_bounded_context(&pool, &bc_name)
.await
.unwrap();
assert_eq!(events.len(), 1);
match &events[0].origin {
skilj_core::event_store::EventOrigin::CommandTriggered { command } => {
assert_eq!(command.command_type.name, "WithdrawMoney");
assert_eq!(command.payload, r#"{"amount":20}"#);
}
other => panic!("expected a CommandTriggered origin, got {other:?}"),
}
});
}
#[test]
fn command_trigger_deduplicates_a_repeated_idempotency_key() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, credential, pool, bc_name, _, _) = setup().await;
let router = skilj.rest_router();
let request = || {
Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {credential}"))
.header("content-type", "application/json")
.header("Idempotency-Key", "retry-1")
.body(Body::from(r#"{"payload":{"amount":20}}"#))
.unwrap()
};
let first_response = router.clone().oneshot(request()).await.unwrap();
assert_eq!(first_response.status(), StatusCode::OK);
let first_body = first_response
.into_body()
.collect()
.await
.unwrap()
.to_bytes();
let first_json: serde_json::Value = serde_json::from_slice(&first_body).unwrap();
assert_eq!(first_json["accepted"], true);
assert_eq!(first_json["deduplicated"], false);
let first_sequences = first_json["triggeredEventSequences"].clone();
let second_response = router.oneshot(request()).await.unwrap();
assert_eq!(second_response.status(), StatusCode::OK);
let second_body = second_response
.into_body()
.collect()
.await
.unwrap()
.to_bytes();
let second_json: serde_json::Value = serde_json::from_slice(&second_body).unwrap();
assert_eq!(second_json["accepted"], true);
assert_eq!(
second_json["deduplicated"], true,
"a repeated Idempotency-Key must be reported as deduplicated"
);
assert_eq!(
second_json["triggeredEventSequences"], first_sequences,
"a dedup hit must return the *original* sequences, not a fresh decision"
);
let events = db::list_events_for_bounded_context(&pool, &bc_name)
.await
.unwrap();
assert_eq!(
events.len(),
1,
"a deduplicated retry must not double-apply the command"
);
});
}
#[test]
fn command_trigger_rejects_a_reserved_idempotency_key_prefix() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, credential, pool, bc_name, _, _) = setup().await;
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {credential}"))
.header("content-type", "application/json")
.header(
"Idempotency-Key",
"skilj-cross-context-route:some-other-route:42",
)
.body(Body::from(r#"{"payload":{"amount":20}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let body = response.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["code"], "reserved_idempotency_key_prefix");
let events = db::list_events_for_bounded_context(&pool, &bc_name)
.await
.unwrap();
assert_eq!(events.len(), 0);
});
}
#[test]
fn command_trigger_rejects_a_payload_that_does_not_match_the_schema_with_400() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, credential, _pool, _bc_name, _, _) = setup().await;
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {credential}"))
.header("content-type", "application/json")
.body(Body::from(r#"{"payload":{"amount":"not a number"}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let body = response.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["code"], "payload_does_not_match_schema");
});
}
#[test]
fn command_trigger_derives_real_tags_from_a_real_tag_mapping() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, credential, pool, bc_name, _, _) = setup().await;
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {credential}"))
.header("content-type", "application/json")
.body(Body::from(r#"{"payload":{"amount":20}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let events = db::list_events_for_bounded_context(&pool, &bc_name)
.await
.unwrap();
assert_eq!(events.len(), 1);
match &events[0].origin {
skilj_core::event_store::EventOrigin::CommandTriggered { command } => {
assert_eq!(
command.consistency_tags,
vec![skilj_core::shared::Tag {
key: "amount".into(),
value: Some("20".into()),
}]
);
}
other => panic!("expected a CommandTriggered origin, got {other:?}"),
}
});
}
#[test]
fn command_trigger_rejects_a_command_whose_owner_does_not_match_the_tokens_scope() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, _credential, pool, bc_name, mapping, command_type) = setup().await;
let scoped_token = access_control::create_command_token(
&mapping,
&command_type,
generate_token_id(),
generate_token_secret(),
Some("20".into()),
test_now(),
)
.unwrap();
db::insert_command_token(&pool, &scoped_token)
.await
.unwrap();
let scoped_credential = format!("{}.{}", scoped_token.id, scoped_token.secret);
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {scoped_credential}"))
.header("content-type", "application/json")
.body(Body::from(r#"{"payload":{"amount":999}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::FORBIDDEN);
let body = response.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["code"], "grant_scope_mismatch");
let events = db::list_events_for_bounded_context(&pool, &bc_name)
.await
.unwrap();
assert!(events.is_empty());
});
}
#[test]
fn command_trigger_succeeds_when_the_owner_matches_the_tokens_scope() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, _credential, pool, bc_name, mapping, command_type) = setup().await;
let scoped_token = access_control::create_command_token(
&mapping,
&command_type,
generate_token_id(),
generate_token_secret(),
Some("20".into()),
test_now(),
)
.unwrap();
db::insert_command_token(&pool, &scoped_token)
.await
.unwrap();
let scoped_credential = format!("{}.{}", scoped_token.id, scoped_token.secret);
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {scoped_credential}"))
.header("content-type", "application/json")
.body(Body::from(r#"{"payload":{"amount":20}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = response.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["accepted"], true);
let events = db::list_events_for_bounded_context(&pool, &bc_name)
.await
.unwrap();
assert_eq!(events.len(), 1);
});
}
#[test]
fn command_trigger_returns_a_rejection_as_200_not_an_error() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, credential, _pool, _bc_name, _, _) = setup().await;
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", format!("Bearer {credential}"))
.header("content-type", "application/json")
.body(Body::from(r#"{"payload":{"amount":5000}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = response.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["accepted"], false);
assert_eq!(json["rejectionKind"], "insufficient_funds");
assert!(json.get("triggeredEventSequences").is_none());
});
}
#[test]
fn command_trigger_rejects_a_malformed_credential() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, _credential, _pool, _bc_name, _, _) = setup().await;
let router = skilj.rest_router();
let request = Request::builder()
.method("POST")
.uri("/v1/commands/trigger")
.header("authorization", "Bearer not-a-real-credential")
.header("content-type", "application/json")
.body(Body::from(r#"{"payload":{"amount":20}}"#))
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
});
}
#[test]
fn command_dispatcher_required_role_reflects_the_requires_role_attribute() {
runtime().block_on(async {
if test_db().await.is_none() {
return;
}
let (skilj, _credential, _pool, bc_name, _, _) = setup().await;
let dispatcher = skilj.command_dispatcher();
assert_eq!(
dispatcher.required_role(&bc_name, "CloseAccount"),
Some(Some("treasury_officer"))
);
assert_eq!(
dispatcher.required_role(&bc_name, "WithdrawMoney"),
Some(None)
);
assert_eq!(dispatcher.required_role(&bc_name, "NoSuchCommand"), None);
});
}