diff options
Diffstat (limited to 'crates/rule-executor/src/processor/rule')
3 files changed, 245 insertions, 2 deletions
diff --git a/crates/rule-executor/src/processor/rule/configuration.rs b/crates/rule-executor/src/processor/rule/configuration.rs index 5f384aa..d8579e6 100644 --- a/crates/rule-executor/src/processor/rule/configuration.rs +++ b/crates/rule-executor/src/processor/rule/configuration.rs @@ -37,10 +37,8 @@ pub(super) async fn get_configuration( .configuration .ok_or_else(|| anyhow!("missing configuration"))?; - println!("inserting"); let cache = state.local_cache.write().await; cache.insert(request, config.clone()).await; - println!("inserted"); Ok(config) } diff --git a/crates/rule-executor/src/processor/rule/determine_outcome.rs b/crates/rule-executor/src/processor/rule/determine_outcome.rs new file mode 100644 index 0000000..727846d --- /dev/null +++ b/crates/rule-executor/src/processor/rule/determine_outcome.rs @@ -0,0 +1,119 @@ +use tracing::trace; +use warden_core::{configuration::rule::Band, message::RuleResult}; + +pub(super) fn determine_outcome(value: i64, bands: &[Band], rule_result: &mut RuleResult) { + trace!("calculating outcome"); + for band in bands { + let value_f64 = value as f64; + + if band.lower_limit.is_none_or(|lower| value_f64 >= lower) + && band.upper_limit.is_none_or(|upper| value_f64 < upper) + { + rule_result.sub_rule_ref = band.sub_rule_ref.to_owned(); + rule_result.reason = band.reason.to_owned(); + break; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_band(lower: Option<f64>, upper: Option<f64>, sub_ref: &str, reason: &str) -> Band { + Band { + lower_limit: lower, + upper_limit: upper, + sub_rule_ref: sub_ref.to_string(), + reason: reason.to_string(), + } + } + + #[test] + fn matches_band_within_limits() { + let bands = vec![ + make_band(Some(0.0), Some(10.0), "A", "Between 0 and 10"), + make_band(Some(10.0), Some(20.0), "B", "Between 10 and 20"), + ]; + let mut rule_result = RuleResult::default(); + + determine_outcome(5, &bands, &mut rule_result); + + assert_eq!(rule_result.sub_rule_ref, "A"); + assert_eq!(rule_result.reason, "Between 0 and 10"); + } + + #[test] + fn matches_band_lower_inclusive_upper_exclusive() { + let bands = vec![ + make_band(Some(0.0), Some(10.0), "A", "Between 0 and 10"), + make_band(Some(10.0), Some(20.0), "B", "Between 10 and 20"), + ]; + let mut rule_result = RuleResult::default(); + + determine_outcome(10, &bands, &mut rule_result); + + assert_eq!(rule_result.sub_rule_ref, "B"); + assert_eq!(rule_result.reason, "Between 10 and 20"); + } + + #[test] + fn no_match_when_above_all_bands() { + let bands = vec![ + make_band(Some(0.0), Some(10.0), "A", "Between 0 and 10"), + make_band(Some(10.0), Some(20.0), "B", "Between 10 and 20"), + ]; + let mut rule_result = RuleResult::default(); + + determine_outcome(30, &bands, &mut rule_result); + + assert_eq!(rule_result, RuleResult::default()); + } + + #[test] + fn match_when_no_upper_limit() { + let bands = vec![make_band(Some(0.0), None, "A", "Above 0")]; + let mut rule_result = RuleResult::default(); + + determine_outcome(100, &bands, &mut rule_result); + + assert_eq!(rule_result.sub_rule_ref, "A"); + assert_eq!(rule_result.reason, "Above 0"); + } + + #[test] + fn match_when_no_lower_limit() { + let bands = vec![make_band(None, Some(50.0), "A", "Below 50")]; + let mut rule_result = RuleResult::default(); + + determine_outcome(-10, &bands, &mut rule_result); + + assert_eq!(rule_result.sub_rule_ref, "A"); + assert_eq!(rule_result.reason, "Below 50"); + } + + #[test] + fn match_when_no_limits() { + let bands = vec![make_band(None, None, "A", "Any value")]; + let mut rule_result = RuleResult::default(); + + determine_outcome(9999, &bands, &mut rule_result); + + assert_eq!(rule_result.sub_rule_ref, "A"); + assert_eq!(rule_result.reason, "Any value"); + } + + #[test] + fn stops_after_first_match() { + let bands = vec![ + make_band(None, None, "A", "Any value"), + make_band(None, None, "B", "Second band"), + ]; + let mut rule_result = RuleResult::default(); + + determine_outcome(5, &bands, &mut rule_result); + + assert_eq!(rule_result.sub_rule_ref, "A"); + assert_eq!(rule_result.reason, "Any value"); + } +} diff --git a/crates/rule-executor/src/processor/rule/rule_901.rs b/crates/rule-executor/src/processor/rule/rule_901.rs new file mode 100644 index 0000000..8dfe036 --- /dev/null +++ b/crates/rule-executor/src/processor/rule/rule_901.rs @@ -0,0 +1,126 @@ +use anyhow::{Result, anyhow}; +use determine_outcome::determine_outcome; +use opentelemetry_semantic_conventions::attribute; +use serde::Deserialize; +use sqlx::types::BigDecimal; +use time::OffsetDateTime; +use tracing::{Instrument, error, info_span, trace}; +use tracing_opentelemetry::OpenTelemetrySpanExt; +use warden_core::{ + configuration::rule::RuleConfiguration, + iso20022::TransactionType, + message::{Payload, RuleResult}, +}; + +use crate::{processor::rule::determine_outcome, state::AppHandle}; + +#[derive(Deserialize)] +pub struct Parameters { + max_query_range: f64, +} + +pub(super) async fn process_901( + configuration: &RuleConfiguration, + payload: &Payload, + state: AppHandle, +) -> Result<RuleResult> { + let mut rule_result = RuleResult { + id: configuration.id.to_string(), + version: configuration.version.to_string(), + ..Default::default() + }; + let c = configuration.configuration.as_ref(); + + let bands = c + .and_then(|value| { + if value.bands.is_empty() { + None + } else { + Some(&value.bands) + } + }) + .ok_or_else(|| anyhow!("no bands available"))?; + + let exit_conditions = c + .and_then(|value| { + if value.exit_conditions.is_empty() { + None + } else { + Some(&value.exit_conditions) + } + }) + .ok_or_else(|| anyhow!("no exit conditions available"))?; + + let parameters = c + .and_then(|value| value.parameters.as_ref()) + .ok_or_else(|| anyhow!("no parameters available"))?; + + let params: Parameters = serde_json::from_value(parameters.clone().into()) + .inspect_err(|e| error!("failed to deserailise params: {e:?}"))?; + + let unsuccessful_transaction = exit_conditions + .iter() + .find(|value| value.sub_rule_ref.eq(".x00")); + + if let Some(warden_core::message::payload::Transaction::Pacs002(pacs002_document)) = + payload.transaction.as_ref() + { + let tx_sts = pacs002_document + .f_i_to_f_i_pmt_sts_rpt + .tx_inf_and_sts + .first() + .ok_or_else(|| anyhow::anyhow!("tx sts to be there"))?; + + if tx_sts.tx_sts().ne("ACCC") { + let unsuccessful_transaction = unsuccessful_transaction + .ok_or_else(|| anyhow::anyhow!("no unsuccessful transaction ref"))?; + rule_result.reason = unsuccessful_transaction.reason.to_owned(); + rule_result.sub_rule_ref = unsuccessful_transaction.sub_rule_ref.to_owned(); + + return Ok(rule_result); + } + + let current_pacs002_timeframe: OffsetDateTime = pacs002_document + .f_i_to_f_i_pmt_sts_rpt + .grp_hdr + .cre_dt_tm + .try_into()?; + + let data_cache = payload + .data_cache + .as_ref() + .ok_or_else(|| anyhow::anyhow!("data cache is missing"))?; + + let range = BigDecimal::try_from(params.max_query_range)?; + + let tx_tp = TransactionType::PACS002.to_string(); + + let span = info_span!("rule.logic"); + span.set_attribute(attribute::DB_SYSTEM_NAME, "postgres"); + span.set_attribute(attribute::DB_OPERATION_NAME, "901"); + span.set_attribute("otel.kind", "client"); + + trace!("executing rule query"); + let recent_transactions = sqlx::query_scalar!( + "select count(*) from transaction_relationship tr + where tr.destination = $1 + and tr.tx_tp = $2 + and extract(epoch from ($3::timestamptz - tr.cre_dt_tm)) * 1000 <= $4 + and tr.cre_dt_tm <= $3::timestamptz", + data_cache.dbtr_acct_id, + tx_tp, + current_pacs002_timeframe, + range, + ) + .fetch_one(&state.services.postgres) + .instrument(span) + .await? + .ok_or_else(|| anyhow::anyhow!("no data"))?; + + determine_outcome(recent_transactions, bands.as_ref(), &mut rule_result); + + Ok(rule_result) + } else { + Err(anyhow::anyhow!("no valid transaction")) + } +} |