feat: rule 901 (#10)HEADmaster

3 files changed, 253 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..e9113e4
--- /dev/null
+++ b/crates/rule-executor/src/processor/rule/determine_outcome.rs
@@ -0,0 +1,127 @@
+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"))
+    }
+}