refactor: add infra-values in intent kafka events (#8264)

crates/common_utils/src/types/keymanager.rs

@@ -49,6 +49,7 @@ pub struct KeyManagerState {
     pub ca: Secret<String>,
     #[cfg(feature = "keymanager_mtls")]
     pub cert: Secret<String>,
+    pub infra_values: Option<serde_json::Value>,
 }
 
 pub trait GetKeymanagerTenant {

crates/router/src/db/kafka_store.rs

@@ -1866,7 +1866,12 @@ impl PaymentIntentInterface for KafkaStore {
 
         if let Err(er) = self
             .kafka_producer
-            .log_payment_intent(&intent, Some(this), self.tenant_id.clone())
+            .log_payment_intent(
+                &intent,
+                Some(this),
+                self.tenant_id.clone(),
+                state.infra_values.clone(),
+            )
             .await
         {
             logger::error!(message="Failed to add analytics entry for Payment Intent {intent:?}", error_message=?er);
@@ -1890,7 +1895,12 @@ impl PaymentIntentInterface for KafkaStore {
 
         if let Err(er) = self
             .kafka_producer
-            .log_payment_intent(&intent, None, self.tenant_id.clone())
+            .log_payment_intent(
+                &intent,
+                None,
+                self.tenant_id.clone(),
+                state.infra_values.clone(),
+            )
             .await
         {
             logger::error!(message="Failed to add analytics entry for Payment Intent {intent:?}", error_message=?er);
@@ -3596,7 +3606,12 @@ impl BatchSampleDataInterface for KafkaStore {
         for payment_intent in payment_intents_list.iter() {
             let _ = self
                 .kafka_producer
-                .log_payment_intent(payment_intent, None, self.tenant_id.clone())
+                .log_payment_intent(
+                    payment_intent,
+                    None,
+                    self.tenant_id.clone(),
+                    state.infra_values.clone(),
+                )
                 .await;
         }
         Ok(payment_intents_list)
@@ -3680,7 +3695,11 @@ impl BatchSampleDataInterface for KafkaStore {
         for payment_intent in payment_intents_list.iter() {
             let _ = self
                 .kafka_producer
-                .log_payment_intent_delete(payment_intent, self.tenant_id.clone())
+                .log_payment_intent_delete(
+                    payment_intent,
+                    self.tenant_id.clone(),
+                    state.infra_values.clone(),
+                )
                 .await;
         }
         Ok(payment_intents_list)

crates/router/src/services/kafka.rs

@@ -473,10 +473,11 @@ impl KafkaProducer {
         intent: &PaymentIntent,
         old_intent: Option<PaymentIntent>,
         tenant_id: TenantID,
+        infra_values: Option<Value>,
     ) -> MQResult<()> {
         if let Some(negative_event) = old_intent {
             self.log_event(&KafkaEvent::old(
-                &KafkaPaymentIntent::from_storage(&negative_event),
+                &KafkaPaymentIntent::from_storage(&negative_event, infra_values.clone()),
                 tenant_id.clone(),
                 self.ckh_database_name.clone(),
             ))
@@ -486,14 +487,14 @@ impl KafkaProducer {
         };
 
         self.log_event(&KafkaEvent::new(
-            &KafkaPaymentIntent::from_storage(intent),
+            &KafkaPaymentIntent::from_storage(intent, infra_values.clone()),
             tenant_id.clone(),
             self.ckh_database_name.clone(),
         ))
         .attach_printable_lazy(|| format!("Failed to add positive intent event {intent:?}"))?;
 
         self.log_event(&KafkaConsolidatedEvent::new(
-            &KafkaPaymentIntentEvent::from_storage(intent),
+            &KafkaPaymentIntentEvent::from_storage(intent, infra_values.clone()),
             tenant_id.clone(),
         ))
         .attach_printable_lazy(|| format!("Failed to add consolidated intent event {intent:?}"))
@@ -503,9 +504,10 @@ impl KafkaProducer {
         &self,
         delete_old_intent: &PaymentIntent,
         tenant_id: TenantID,
+        infra_values: Option<Value>,
     ) -> MQResult<()> {
         self.log_event(&KafkaEvent::old(
-            &KafkaPaymentIntent::from_storage(delete_old_intent),
+            &KafkaPaymentIntent::from_storage(delete_old_intent, infra_values),
             tenant_id.clone(),
             self.ckh_database_name.clone(),
         ))

crates/router/src/services/kafka/payment_intent.rs

@@ -42,6 +42,8 @@ pub struct KafkaPaymentIntent<'a> {
     pub feature_metadata: Option<&'a Value>,
     pub merchant_order_reference_id: Option<&'a String>,
     pub organization_id: &'a id_type::OrganizationId,
+    #[serde(flatten)]
+    infra_values: Option<Value>,
 }
 
 #[cfg(feature = "v2")]
@@ -81,7 +83,7 @@ pub struct KafkaPaymentIntent<'a> {
 
 #[cfg(feature = "v1")]
 impl<'a> KafkaPaymentIntent<'a> {
-    pub fn from_storage(intent: &'a PaymentIntent) -> Self {
+    pub fn from_storage(intent: &'a PaymentIntent, infra_values: Option<Value>) -> Self {
         Self {
             payment_id: &intent.payment_id,
             merchant_id: &intent.merchant_id,
@@ -121,13 +123,14 @@ impl<'a> KafkaPaymentIntent<'a> {
             feature_metadata: intent.feature_metadata.as_ref(),
             merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
             organization_id: &intent.organization_id,
+            infra_values,
         }
     }
 }
 
 #[cfg(feature = "v2")]
 impl<'a> KafkaPaymentIntent<'a> {
-    pub fn from_storage(intent: &'a PaymentIntent) -> Self {
+    pub fn from_storage(intent: &'a PaymentIntent, infra_values: Option<Value>) -> Self {
         // Self {
         //     id: &intent.id,
         //     merchant_id: &intent.merchant_id,

crates/router/src/services/kafka/payment_intent_event.rs

@@ -43,6 +43,8 @@ pub struct KafkaPaymentIntentEvent<'a> {
     pub feature_metadata: Option<&'a Value>,
     pub merchant_order_reference_id: Option<&'a String>,
     pub organization_id: &'a id_type::OrganizationId,
+    #[serde(flatten)]
+    pub infra_values: Option<Value>,
 }
 
 #[cfg(feature = "v2")]
@@ -95,7 +97,7 @@ impl KafkaPaymentIntentEvent<'_> {
 
 #[cfg(feature = "v1")]
 impl<'a> KafkaPaymentIntentEvent<'a> {
-    pub fn from_storage(intent: &'a PaymentIntent) -> Self {
+    pub fn from_storage(intent: &'a PaymentIntent, infra_values: Option<Value>) -> Self {
         Self {
             payment_id: &intent.payment_id,
             merchant_id: &intent.merchant_id,
@@ -135,13 +137,14 @@ impl<'a> KafkaPaymentIntentEvent<'a> {
             feature_metadata: intent.feature_metadata.as_ref(),
             merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
             organization_id: &intent.organization_id,
+            infra_values: infra_values.clone(),
         }
     }
 }
 
 #[cfg(feature = "v2")]
 impl<'a> KafkaPaymentIntentEvent<'a> {
-    pub fn from_storage(intent: &'a PaymentIntent) -> Self {
+    pub fn from_storage(intent: &'a PaymentIntent, infra_values: Option<Value>) -> Self {
         // Self {
         //     id: &intent.id,
         //     merchant_id: &intent.merchant_id,

crates/router/src/types/domain/types.rs

@@ -21,6 +21,7 @@ impl From<&app::SessionState> for KeyManagerState {
             cert: conf.cert.clone(),
             #[cfg(feature = "keymanager_mtls")]
             ca: conf.ca.clone(),
+            infra_values: app::AppState::process_env_mappings(state.conf.infra_values.clone()),
         }
     }
 }