feat(revenue_recovery): Add token active status filtering and account update history tracking (#10056)

Co-authored-by: sagarnaikjuspay <sagar.naik@juspay.net>

crates/api_models/src/revenue_recovery_data_backfill.rs

@@ -9,6 +9,8 @@ use masking::Secret;
 use serde::{Deserialize, Serialize};
 use time::{Date, PrimitiveDateTime};
 
+use crate::payments;
+
 #[derive(Debug, Deserialize, Serialize)]
 pub struct RevenueRecoveryBackfillRequest {
     pub bin_number: Option<Secret<String>>,
@@ -21,6 +23,30 @@ pub struct RevenueRecoveryBackfillRequest {
     pub clean_bank_name: Option<String>,
     pub country_name: Option<String>,
     pub daily_retry_history: Option<String>,
+    pub is_active: Option<bool>,
+    #[serde(
+        default,
+        deserialize_with = "RevenueRecoveryBackfillRequest::deserialize_history_vec_opt"
+    )]
+    pub account_update_history: Option<Vec<AccountUpdateHistoryRecord>>,
+}
+
+impl RevenueRecoveryBackfillRequest {
+    pub fn deserialize_history_vec_opt<'de, D>(
+        deserializer: D,
+    ) -> Result<Option<Vec<AccountUpdateHistoryRecord>>, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        use serde::Deserialize;
+        let val = Option::<String>::deserialize(deserializer)?;
+        match val.as_deref().map(str::trim) {
+            None | Some("") => Ok(None),
+            Some(s) => serde_json::from_str::<Vec<AccountUpdateHistoryRecord>>(s)
+                .map(Some)
+                .map_err(serde::de::Error::custom),
+        }
+    }
 }
 
 #[derive(Debug, Serialize)]
@@ -46,6 +72,16 @@ pub struct CsvParsingError {
     pub error: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AccountUpdateHistoryRecord {
+    pub old_token: String,
+    pub new_token: String,
+    #[serde(with = "common_utils::custom_serde::iso8601")]
+    pub updated_at: PrimitiveDateTime,
+    pub old_token_info: Option<payments::AdditionalCardInfo>,
+    pub new_token_info: Option<payments::AdditionalCardInfo>,
+}
+
 /// Comprehensive card
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ComprehensiveCardData {
@@ -56,6 +92,8 @@ pub struct ComprehensiveCardData {
     pub card_issuer: Option<String>,
     pub card_issuing_country: Option<String>,
     pub daily_retry_history: Option<HashMap<Date, i32>>,
+    pub is_active: Option<bool>,
+    pub account_update_history: Option<Vec<AccountUpdateHistoryRecord>>,
 }
 
 impl ApiEventMetric for RevenueRecoveryDataBackfillResponse {

crates/router/src/core/revenue_recovery_data_backfill.rs

@@ -255,6 +255,8 @@ async fn process_payment_method_record(
                 card_issuer: None,
                 card_issuing_country: None,
                 daily_retry_history: None,
+                is_active: None,
+                account_update_history: None,
             }
         }
     };
@@ -401,6 +403,8 @@ fn build_comprehensive_card_data(
         card_issuer,
         card_issuing_country,
         daily_retry_history,
+        is_active: record.is_active,
+        account_update_history: record.account_update_history.clone(),
     })
 }
 

crates/router/src/core/webhooks/recovery_incoming.rs

@@ -957,6 +957,8 @@ impl RevenueRecoveryAttempt {
                 card_network: revenue_recovery_attempt_data.card_info.card_network.clone(),
                 card_type: revenue_recovery_attempt_data.card_info.card_type.clone(),
             },
+            is_active: Some(true), // Tokens created from recovery attempts are active by default
+            account_update_history: None, // No prior account update history exists for freshly ingested tokens
         };
 
         // Make the Redis call to store tokens

crates/router/src/types/storage/revenue_recovery_redis_operation.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use api_models::revenue_recovery_data_backfill::{self, RedisKeyType};
+use api_models::revenue_recovery_data_backfill::{self, AccountUpdateHistoryRecord, RedisKeyType};
 use common_enums::enums::CardNetwork;
 use common_utils::{date_time, errors::CustomResult, id_type};
 use error_stack::ResultExt;
@@ -45,6 +45,10 @@ pub struct PaymentProcessorTokenStatus {
     pub is_hard_decline: Option<bool>,
     /// Timestamp of the last modification to this token status
     pub modified_at: Option<PrimitiveDateTime>,
+    /// Indicates if the token is active or not
+    pub is_active: Option<bool>,
+    /// Update history of the token
+    pub account_update_history: Option<Vec<AccountUpdateHistoryRecord>>,
 }
 
 /// Token retry availability information with detailed wait times
@@ -556,6 +560,8 @@ impl RedisTokenManager {
                             OffsetDateTime::now_utc().date(),
                             OffsetDateTime::now_utc().time(),
                         )),
+                        is_active: status.is_active,
+                        account_update_history: status.account_update_history.clone(),
                     })
             }
             None => None,
@@ -632,6 +638,8 @@ impl RedisTokenManager {
                     OffsetDateTime::now_utc().date(),
                     OffsetDateTime::now_utc().time(),
                 )),
+                is_active: status.is_active,
+                account_update_history: status.account_update_history.clone(),
             };
             updated_tokens_map.insert(token_id, updated_status);
         }
@@ -680,6 +688,8 @@ impl RedisTokenManager {
                         OffsetDateTime::now_utc().date(),
                         OffsetDateTime::now_utc().time(),
                     )),
+                    is_active: status.is_active,
+                    account_update_history: status.account_update_history.clone(),
                 });
 
         match updated_token {
@@ -1007,6 +1017,27 @@ impl RedisTokenManager {
             OffsetDateTime::now_utc().time(),
         ));
 
+        // Update account_update_history if provided
+        if let Some(history) = &card_data.account_update_history {
+            // Convert api_models::AccountUpdateHistoryRecord to storage::AccountUpdateHistoryRecord
+            let converted_history: Vec<AccountUpdateHistoryRecord> = history
+                .iter()
+                .map(|api_record| AccountUpdateHistoryRecord {
+                    old_token: api_record.old_token.clone(),
+                    new_token: api_record.new_token.clone(),
+                    updated_at: api_record.updated_at,
+                    old_token_info: api_record.old_token_info.clone(),
+                    new_token_info: api_record.new_token_info.clone(),
+                })
+                .collect();
+            existing_token.account_update_history = Some(converted_history);
+        }
+
+        // Update is_active if provided
+        card_data.is_active.map(|is_active| {
+            existing_token.is_active = Some(is_active);
+        });
+
         // Save the updated token map back to Redis
         Self::update_or_add_connector_customer_payment_processor_tokens(
             state,

crates/router/src/workflows/revenue_recovery.rs

@@ -768,7 +768,12 @@ pub async fn get_best_psp_token_available_for_smart_retry(
                     errors::RedisError::RedisConnectionError.into(),
                 ))?;
 
-            let result = RedisTokenManager::get_tokens_with_retry_metadata(state, &existing_tokens);
+            let active_tokens: HashMap<_, _> = existing_tokens
+                .into_iter()
+                .filter(|(_, token_status)| token_status.is_active != Some(false))
+                .collect();
+
+            let result = RedisTokenManager::get_tokens_with_retry_metadata(state, &active_tokens);
 
             let payment_processor_token_response =
                 call_decider_for_payment_processor_tokens_select_closest_time(