rust/hyperswitch
1
0
Fork 0
mirror of https://github.com/juspay/hyperswitch.git synced 2025-10-28 04:04:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/juspay/hyperswitch into ucs-authorizedotnet-fix

Browse Source
This commit is contained in:
Amitsingh Tanwar
2025-10-25 11:27:37 +05:30
parent b4224998be d62eed7c80
commit df3796f20d
6 changed files with 384 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
crates/router/src/core/revenue_recovery.rs
View File

@ -529,7 +529,7 @@ pub async fn perform_calculate_workflow(
.await?;
// 2. Get best available token
let best_time_to_schedule =
let payment_processor_token_response =
match revenue_recovery_workflow::get_token_with_schedule_time_based_on_retry_algorithm_type(
state,
&connector_customer_id,
@ -546,12 +546,14 @@ pub async fn perform_calculate_workflow(
connector_customer_id = %connector_customer_id,
"Failed to get best PSP token"
);
None
revenue_recovery_workflow::PaymentProcessorTokenResponse::None
}
};
match best_time_to_schedule {
Some(scheduled_time) => {
match payment_processor_token_response {
revenue_recovery_workflow::PaymentProcessorTokenResponse::ScheduledTime {
scheduled_time,
} => {
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
@ -602,113 +604,86 @@ pub async fn perform_calculate_workflow(
);
}
None => {
let scheduled_token = match storage::revenue_recovery_redis_operation::
RedisTokenManager::get_payment_processor_token_with_schedule_time(state, &connector_customer_id)
.await {
Ok(scheduled_token_opt) => scheduled_token_opt,
Err(e) => {
logger::error!(
error = ?e,
connector_customer_id = %connector_customer_id,
"Failed to get PSP token status"
);
None
}
};
revenue_recovery_workflow::PaymentProcessorTokenResponse::NextAvailableTime {
next_available_time,
} => {
// Update scheduled time to next_available_time + Buffer
// here next_available_time is the wait time
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"No token but time available, rescheduling for scheduled time "
);
match scheduled_token {
Some(scheduled_token) => {
// Update scheduled time to scheduled time + 15 minutes
// here scheduled_time is the wait time 15 minutes is a buffer time that we are adding
logger::info!(
update_calculate_job_schedule_time(
db,
process,
time::Duration::seconds(
state
.conf
.revenue_recovery
.recovery_timestamp
.job_schedule_buffer_time_in_seconds,
),
Some(next_available_time),
&connector_customer_id,
retry_algorithm_type,
)
.await?;
}
revenue_recovery_workflow::PaymentProcessorTokenResponse::None => {
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"Hard decline flag is false, rescheduling for scheduled time + 15 mins"
);
update_calculate_job_schedule_time(
db,
process,
time::Duration::seconds(
state
.conf
.revenue_recovery
.recovery_timestamp
.job_schedule_buffer_time_in_seconds,
),
Some(common_utils::date_time::now()),
&connector_customer_id,
retry_algorithm_type,
)
.await?;
}
revenue_recovery_workflow::PaymentProcessorTokenResponse::HardDecline => {
// Finish calculate workflow with CALCULATE_WORKFLOW_FINISH
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"Token/Tokens is/are Hard decline, finishing CALCULATE_WORKFLOW"
);
db.as_scheduler()
.finish_process_with_business_status(
process.clone(),
business_status::CALCULATE_WORKFLOW_FINISH,
)
.await
.map_err(|e| {
logger::error!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"No token but time available, rescheduling for scheduled time + 15 mins"
error = ?e,
"Failed to finish CALCULATE_WORKFLOW"
);
sch_errors::ProcessTrackerError::ProcessUpdateFailed
})?;
update_calculate_job_schedule_time(
db,
process,
time::Duration::seconds(
state
.conf
.revenue_recovery
.recovery_timestamp
.job_schedule_buffer_time_in_seconds,
),
scheduled_token.scheduled_at,
&connector_customer_id,
)
.await?;
}
None => {
let hard_decline_flag = storage::revenue_recovery_redis_operation::
RedisTokenManager::are_all_tokens_hard_declined(
state,
&connector_customer_id
)
.await
.ok()
.unwrap_or(false);
event_type = Some(common_enums::EventType::PaymentFailed);
match hard_decline_flag {
false => {
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"Hard decline flag is false, rescheduling for scheduled time + 15 mins"
);
update_calculate_job_schedule_time(
db,
process,
time::Duration::seconds(
state
.conf
.revenue_recovery
.recovery_timestamp
.job_schedule_buffer_time_in_seconds,
),
Some(common_utils::date_time::now()),
&connector_customer_id,
)
.await?;
}
true => {
// Finish calculate workflow with CALCULATE_WORKFLOW_FINISH
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"No token available, finishing CALCULATE_WORKFLOW"
);
db.as_scheduler()
.finish_process_with_business_status(
process.clone(),
business_status::CALCULATE_WORKFLOW_FINISH,
)
.await
.map_err(|e| {
logger::error!(
process_id = %process.id,
error = ?e,
"Failed to finish CALCULATE_WORKFLOW"
);
sch_errors::ProcessTrackerError::ProcessUpdateFailed
})?;
event_type = Some(common_enums::EventType::PaymentFailed);
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"CALCULATE_WORKFLOW finished successfully"
);
}
}
}
}
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"CALCULATE_WORKFLOW finished successfully"
);
}
}
@ -749,6 +724,7 @@ async fn update_calculate_job_schedule_time(
additional_time: time::Duration,
base_time: Option<time::PrimitiveDateTime>,
connector_customer_id: &str,
retry_algorithm_type: common_enums::RevenueRecoveryAlgorithmType,
) -> Result<(), sch_errors::ProcessTrackerError> {
let now = common_utils::date_time::now();
@ -759,11 +735,22 @@ async fn update_calculate_job_schedule_time(
connector_customer_id = %connector_customer_id,
"Rescheduling Calculate Job at "
);
let mut old_tracking_data: pcr::RevenueRecoveryWorkflowTrackingData =
serde_json::from_value(process.tracking_data.clone())
.change_context(errors::RecoveryError::ValueNotFound)
.attach_printable("Failed to deserialize the tracking data from process tracker")?;
old_tracking_data.revenue_recovery_retry = retry_algorithm_type;
let tracking_data = serde_json::to_value(old_tracking_data)
.change_context(errors::RecoveryError::ValueNotFound)
.attach_printable("Failed to serialize the tracking data for process tracker")?;
let pt_update = storage::ProcessTrackerUpdate::Update {
name: Some("CALCULATE_WORKFLOW".to_string()),
retry_count: Some(process.clone().retry_count),
schedule_time: Some(new_schedule_time),
tracking_data: Some(process.clone().tracking_data),
tracking_data: Some(tracking_data),
business_status: Some(String::from(business_status::PENDING)),
status: Some(common_enums::ProcessTrackerStatus::Pending),
updated_at: Some(common_utils::date_time::now()),

33
crates/router/src/core/revenue_recovery/types.rs
View File

@ -176,8 +176,7 @@ impl RevenueRecoveryPaymentsAttemptStatus {
state,
&connector_customer_id,
&None,
// Since this is succeeded payment attempt, 'is_hard_decine' will be false.
&Some(false),
&None,
used_token.as_deref(),
)
.await;
@ -493,19 +492,12 @@ impl Action {
);
};
let is_hard_decline = revenue_recovery::check_hard_decline(
state,
&payment_data.payment_attempt,
)
.await
.ok();
// update the status of token in redis
let _update_error_code = storage::revenue_recovery_redis_operation::RedisTokenManager::update_payment_processor_token_error_code_from_process_tracker(
state,
&connector_customer_id,
&None,
&is_hard_decline,
&None,
Some(&scheduled_token.payment_processor_token_details.payment_processor_token),
)
.await;
@ -659,7 +651,7 @@ impl Action {
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"No token available, finishing CALCULATE_WORKFLOW"
"No token available, finishing EXECUTE_WORKFLOW"
);
state
@ -671,12 +663,12 @@ impl Action {
)
.await
.change_context(errors::RecoveryError::ProcessTrackerFailure)
.attach_printable("Failed to finish CALCULATE_WORKFLOW")?;
.attach_printable("Failed to finish EXECUTE_WORKFLOW")?;
logger::info!(
process_id = %process.id,
connector_customer_id = %connector_customer_id,
"CALCULATE_WORKFLOW finished successfully"
"EXECUTE_WORKFLOW finished successfully"
);
Ok(Self::TerminalFailure(payment_attempt.clone()))
}
@ -854,8 +846,7 @@ impl Action {
state,
&connector_customer_id,
&None,
// Since this is succeeded, 'hard_decine' will be false.
&Some(false),
&None,
used_token.as_deref(),
)
.await;
@ -1144,9 +1135,19 @@ pub async fn reopen_calculate_workflow_on_payment_failure(
.change_context(errors::RecoveryError::ValueNotFound)
.attach_printable("Failed to deserialize the tracking data from process tracker")?;
let retry_algorithm_type = profile
.revenue_recovery_retry_algorithm_type
.filter(|retry_type| *retry_type != common_enums::RevenueRecoveryAlgorithmType::Monitoring) // ignore Monitoring
.unwrap_or(old_tracking_data.revenue_recovery_retry);
let new_tracking_data = pcr::RevenueRecoveryWorkflowTrackingData {
payment_attempt_id: latest_attempt_id.clone(),
..old_tracking_data
revenue_recovery_retry: retry_algorithm_type,
merchant_id: old_tracking_data.merchant_id.clone(),
profile_id: old_tracking_data.profile_id.clone(),
global_payment_id: old_tracking_data.global_payment_id.clone(),
billing_mca_id: old_tracking_data.billing_mca_id.clone(),
invoice_scheduled_time: old_tracking_data.invoice_scheduled_time,
};
let tracking_data = serde_json::to_value(new_tracking_data)

3
crates/router/src/core/webhooks/recovery_incoming.rs
View File

@ -908,7 +908,7 @@ impl RevenueRecoveryAttempt {
payment_connector_name: Option<common_enums::connector_enums::Connector>,
) -> CustomResult<(), errors::RevenueRecoveryError> {
let revenue_recovery_attempt_data = &self.0;
let error_code = revenue_recovery_attempt_data.error_code.clone();
let error_code = recovery_attempt.error_code.clone();
let error_message = revenue_recovery_attempt_data.error_message.clone();
let connector_name = payment_connector_name
.ok_or(errors::RevenueRecoveryError::TransactionWebhookProcessingFailed)
@ -939,6 +939,7 @@ impl RevenueRecoveryAttempt {
daily_retry_history: HashMap::from([(recovery_attempt.created_at.date(), 1)]),
scheduled_at: None,
is_hard_decline: Some(is_hard_decline),
modified_at: Some(recovery_attempt.created_at),
payment_processor_token_details: PaymentProcessorTokenDetails {
payment_processor_token: revenue_recovery_attempt_data
.processor_payment_method_token

54
crates/router/src/types/storage/revenue_recovery_redis_operation.rs
View File

@ -43,6 +43,8 @@ pub struct PaymentProcessorTokenStatus {
pub scheduled_at: Option<PrimitiveDateTime>,
/// Indicates if the token is a hard decline (no retries allowed)
pub is_hard_decline: Option<bool>,
/// Timestamp of the last modification to this token status
pub modified_at: Option<PrimitiveDateTime>,
}
/// Token retry availability information with detailed wait times
@ -415,11 +417,19 @@ impl RedisTokenManager {
let monthly_wait_hours =
if total_30_day_retries >= card_network_config.max_retry_count_for_thirty_day {
let mut accumulated_retries = 0;
// Iterate from most recent to oldest
(0..RETRY_WINDOW_DAYS)
.rev()
.map(|i| today - Duration::days(i.into()))
.find(|date| token.daily_retry_history.get(date).copied().unwrap_or(0) > 0)
.map(|date| Self::calculate_wait_hours(date + Duration::days(31), now))
.map(|days_ago| today - Duration::days(days_ago.into()))
.find(|date| {
let retries = token.daily_retry_history.get(date).copied().unwrap_or(0);
accumulated_retries += retries;
accumulated_retries >= card_network_config.max_retry_count_for_thirty_day
})
.map(|breach_date| {
Self::calculate_wait_hours(breach_date + Duration::days(31), now)
})
.unwrap_or(0)
} else {
0
@ -463,13 +473,14 @@ impl RedisTokenManager {
let was_existing = token_map.contains_key(&token_id);
let error_code = token_data.error_code.clone();
let modified_at = token_data.modified_at;
let today = OffsetDateTime::now_utc().date();
token_map
.get_mut(&token_id)
.map(|existing_token| {
error_code.map(|err| existing_token.error_code = Some(err));
Self::normalize_retry_window(existing_token, today);
for (date, &value) in &token_data.daily_retry_history {
@ -479,6 +490,12 @@ impl RedisTokenManager {
.and_modify(|v| *v += value)
.or_insert(value);
}
(existing_token.modified_at < modified_at).then(|| {
existing_token.modified_at = modified_at;
error_code.map(|err| existing_token.error_code = Some(err));
existing_token.is_hard_decline = token_data.is_hard_decline;
});
})
.or_else(|| {
token_map.insert(token_id.clone(), token_data);
@ -529,6 +546,10 @@ impl RedisTokenManager {
daily_retry_history: status.daily_retry_history.clone(),
scheduled_at: None,
is_hard_decline: *is_hard_decline,
modified_at: Some(PrimitiveDateTime::new(
OffsetDateTime::now_utc().date(),
OffsetDateTime::now_utc().time(),
)),
})
}
None => None,
@ -606,6 +627,10 @@ impl RedisTokenManager {
daily_retry_history: status.daily_retry_history.clone(),
scheduled_at: schedule_time,
is_hard_decline: status.is_hard_decline,
modified_at: Some(PrimitiveDateTime::new(
OffsetDateTime::now_utc().date(),
OffsetDateTime::now_utc().time(),
)),
});
match updated_token {
@ -905,6 +930,11 @@ impl RedisTokenManager {
.unwrap_or(Some(existing_scheduled_at)) // No cutoff provided, keep existing value
});
existing_token.modified_at = Some(PrimitiveDateTime::new(
OffsetDateTime::now_utc().date(),
OffsetDateTime::now_utc().time(),
));
// Save the updated token map back to Redis
Self::update_or_add_connector_customer_payment_processor_tokens(
state,
@ -920,4 +950,16 @@ impl RedisTokenManager {
Ok(())
}
pub async fn get_payment_processor_metadata_for_connector_customer(
state: &SessionState,
customer_id: &str,
) -> CustomResult<HashMap<String, PaymentProcessorTokenWithRetryInfo>, errors::StorageError>
{
let token_map =
Self::get_connector_customer_payment_processor_tokens(state, customer_id).await?;
let token_data = Self::get_tokens_with_retry_metadata(state, &token_map);
Ok(token_data)
}
}

2
crates/router/src/workflows/payment_sync.rs
View File

@ -329,7 +329,7 @@ pub async fn recovery_retry_sync_task(
connector_customer_id
.async_map(|id| async move {
let _ = update_token_expiry_based_on_schedule_time(state, &id, Some(s_time))
let _ = update_token_expiry_based_on_schedule_time(state, &id, s_time)
.await
.map_err(|e| {
logger::error!(

295
crates/router/src/workflows/revenue_recovery.rs
View File

@ -525,7 +525,7 @@ pub fn calculate_difference_in_seconds(scheduled_time: time::PrimitiveDateTime)
pub async fn update_token_expiry_based_on_schedule_time(
state: &SessionState,
connector_customer_id: &str,
delayed_schedule_time: Option<time::PrimitiveDateTime>,
delayed_schedule_time: time::PrimitiveDateTime,
) -> CustomResult<(), errors::ProcessTrackerError> {
let expiry_buffer = state
.conf
@ -533,25 +533,40 @@ pub async fn update_token_expiry_based_on_schedule_time(
.recovery_timestamp
.redis_ttl_buffer_in_seconds;
delayed_schedule_time
.async_map(|t| async move {
let expiry_time = calculate_difference_in_seconds(t) + expiry_buffer;
RedisTokenManager::update_connector_customer_lock_ttl(
state,
connector_customer_id,
expiry_time,
)
.await
.change_context(errors::ProcessTrackerError::ERedisError(
errors::RedisError::RedisConnectionError.into(),
))
})
.await
.transpose()?;
let expiry_time = calculate_difference_in_seconds(delayed_schedule_time) + expiry_buffer;
RedisTokenManager::update_connector_customer_lock_ttl(
state,
connector_customer_id,
expiry_time,
)
.await
.change_context(errors::ProcessTrackerError::ERedisError(
errors::RedisError::RedisConnectionError.into(),
));
Ok(())
}
#[cfg(feature = "v2")]
#[derive(Debug)]
pub enum PaymentProcessorTokenResponse {
/// Token HardDecline
HardDecline,
/// Token can be retried at this specific time
ScheduledTime {
scheduled_time: time::PrimitiveDateTime,
},
/// Token locked or unavailable, next attempt possible
NextAvailableTime {
next_available_time: time::PrimitiveDateTime,
},
/// No retry info available / nothing to do yet
None,
}
#[cfg(feature = "v2")]
pub async fn get_token_with_schedule_time_based_on_retry_algorithm_type(
state: &SessionState,
@ -559,9 +574,8 @@ pub async fn get_token_with_schedule_time_based_on_retry_algorithm_type(
payment_intent: &PaymentIntent,
retry_algorithm_type: RevenueRecoveryAlgorithmType,
retry_count: i32,
) -> CustomResult<Option<time::PrimitiveDateTime>, errors::ProcessTrackerError> {
let mut scheduled_time = None;
) -> CustomResult<PaymentProcessorTokenResponse, errors::ProcessTrackerError> {
let mut payment_processor_token_response = PaymentProcessorTokenResponse::None;
match retry_algorithm_type {
RevenueRecoveryAlgorithmType::Monitoring => {
logger::error!("Monitoring type found for Revenue Recovery retry payment");
@ -576,11 +590,67 @@ pub async fn get_token_with_schedule_time_based_on_retry_algorithm_type(
.await
.ok_or(errors::ProcessTrackerError::EApiErrorResponse)?;
scheduled_time = Some(time);
let payment_processor_token = payment_intent
.feature_metadata
.as_ref()
.and_then(|metadata| metadata.payment_revenue_recovery_metadata.as_ref())
.map(|recovery_metadata| {
recovery_metadata
.billing_connector_payment_details
.payment_processor_token
.clone()
});
let payment_processor_tokens_details =
RedisTokenManager::get_payment_processor_metadata_for_connector_customer(
state,
connector_customer_id,
)
.await
.change_context(errors::ProcessTrackerError::ERedisError(
errors::RedisError::RedisConnectionError.into(),
))?;
// Get the token info from redis
let payment_processor_tokens_details_with_retry_info = payment_processor_token
.as_ref()
.and_then(|t| payment_processor_tokens_details.get(t));
// If payment_processor_tokens_details_with_retry_info is None, then no schedule time
match payment_processor_tokens_details_with_retry_info {
None => {
payment_processor_token_response = PaymentProcessorTokenResponse::None;
logger::debug!("No payment processor token found for cascading retry");
}
Some(payment_token) => {
if payment_token.token_status.is_hard_decline.unwrap_or(false) {
payment_processor_token_response =
PaymentProcessorTokenResponse::HardDecline;
} else if payment_token.retry_wait_time_hours > 0 {
let utc_schedule_time: time::OffsetDateTime =
time::OffsetDateTime::now_utc()
+ time::Duration::hours(payment_token.retry_wait_time_hours);
let next_available_time = time::PrimitiveDateTime::new(
utc_schedule_time.date(),
utc_schedule_time.time(),
);
payment_processor_token_response =
PaymentProcessorTokenResponse::NextAvailableTime {
next_available_time,
};
} else {
payment_processor_token_response =
PaymentProcessorTokenResponse::ScheduledTime {
scheduled_time: time,
};
}
}
}
}
RevenueRecoveryAlgorithmType::Smart => {
scheduled_time = get_best_psp_token_available_for_smart_retry(
payment_processor_token_response = get_best_psp_token_available_for_smart_retry(
state,
connector_customer_id,
payment_intent,
@ -589,17 +659,40 @@ pub async fn get_token_with_schedule_time_based_on_retry_algorithm_type(
.change_context(errors::ProcessTrackerError::EApiErrorResponse)?;
}
}
let delayed_schedule_time =
scheduled_time.map(|time| add_random_delay_to_schedule_time(state, time));
let _ = update_token_expiry_based_on_schedule_time(
state,
connector_customer_id,
delayed_schedule_time,
)
.await;
match &mut payment_processor_token_response {
PaymentProcessorTokenResponse::HardDecline => {
logger::debug!("Token is hard declined");
}
Ok(delayed_schedule_time)
PaymentProcessorTokenResponse::ScheduledTime { scheduled_time } => {
// Add random delay to schedule time
*scheduled_time = add_random_delay_to_schedule_time(state, *scheduled_time);
// Log the scheduled retry time at debug level
logger::info!("Retry scheduled at {:?}", scheduled_time);
// Update token expiry based on schedule time
update_token_expiry_based_on_schedule_time(
state,
connector_customer_id,
*scheduled_time,
)
.await;
}
PaymentProcessorTokenResponse::NextAvailableTime {
next_available_time,
} => {
logger::info!("Next available retry at {:?}", next_available_time);
}
PaymentProcessorTokenResponse::None => {
logger::debug!("No retry info available");
}
}
Ok(payment_processor_token_response)
}
#[cfg(feature = "v2")]
@ -607,7 +700,7 @@ pub async fn get_best_psp_token_available_for_smart_retry(
state: &SessionState,
connector_customer_id: &str,
payment_intent: &PaymentIntent,
) -> CustomResult<Option<time::PrimitiveDateTime>, errors::ProcessTrackerError> {
) -> CustomResult<PaymentProcessorTokenResponse, errors::ProcessTrackerError> {
// Lock using payment_id
let locked = RedisTokenManager::lock_connector_customer_status(
state,
@ -619,10 +712,48 @@ pub async fn get_best_psp_token_available_for_smart_retry(
errors::RedisError::RedisConnectionError.into(),
))?;
match !locked {
true => Ok(None),
match locked {
false => {
let token_details =
RedisTokenManager::get_payment_processor_metadata_for_connector_customer(
state,
connector_customer_id,
)
.await
.change_context(errors::ProcessTrackerError::ERedisError(
errors::RedisError::RedisConnectionError.into(),
))?;
// Check token with schedule time in Redis
let token_info_with_schedule_time = token_details
.values()
.find(|info| info.token_status.scheduled_at.is_some());
// Check for hard decline if info is none
let hard_decline_status = token_details
.values()
.all(|token| token.token_status.is_hard_decline.unwrap_or(false));
let mut payment_processor_token_response = PaymentProcessorTokenResponse::None;
if hard_decline_status {
payment_processor_token_response = PaymentProcessorTokenResponse::HardDecline;
} else {
payment_processor_token_response = match token_info_with_schedule_time
.as_ref()
.and_then(|t| t.token_status.scheduled_at)
{
Some(scheduled_time) => PaymentProcessorTokenResponse::NextAvailableTime {
next_available_time: scheduled_time,
},
None => PaymentProcessorTokenResponse::None,
};
}
Ok(payment_processor_token_response)
}
true => {
// Get existing tokens from Redis
let existing_tokens =
RedisTokenManager::get_connector_customer_payment_processor_tokens(
@ -634,20 +765,19 @@ pub async fn get_best_psp_token_available_for_smart_retry(
errors::RedisError::RedisConnectionError.into(),
))?;
// TODO: Insert into payment_intent_feature_metadata (DB operation)
let result = RedisTokenManager::get_tokens_with_retry_metadata(state, &existing_tokens);
let best_token_time = call_decider_for_payment_processor_tokens_select_closet_time(
state,
&result,
payment_intent,
connector_customer_id,
)
.await
.change_context(errors::ProcessTrackerError::EApiErrorResponse)?;
let payment_processor_token_response =
call_decider_for_payment_processor_tokens_select_closest_time(
state,
&result,
payment_intent,
connector_customer_id,
)
.await
.change_context(errors::ProcessTrackerError::EApiErrorResponse)?;
Ok(best_token_time)
Ok(payment_processor_token_response)
}
}
}
@ -709,40 +839,42 @@ async fn process_token_for_retry(
#[cfg(feature = "v2")]
#[allow(clippy::too_many_arguments)]
pub async fn call_decider_for_payment_processor_tokens_select_closet_time(
pub async fn call_decider_for_payment_processor_tokens_select_closest_time(
state: &SessionState,
processor_tokens: &HashMap<String, PaymentProcessorTokenWithRetryInfo>,
payment_intent: &PaymentIntent,
connector_customer_id: &str,
) -> CustomResult<Option<time::PrimitiveDateTime>, errors::ProcessTrackerError> {
tracing::debug!("Filtered payment attempts based on payment tokens",);
) -> CustomResult<PaymentProcessorTokenResponse, errors::ProcessTrackerError> {
let mut tokens_with_schedule_time: Vec<ScheduledToken> = Vec::new();
for token_with_retry_info in processor_tokens.values() {
let token_details = &token_with_retry_info
.token_status
.payment_processor_token_details;
let error_code = token_with_retry_info.token_status.error_code.clone();
// Check for successful token
let mut token_with_none_error_code = processor_tokens.values().find(|token| {
token.token_status.error_code.is_none()
&& !token.token_status.is_hard_decline.unwrap_or(false)
});
match error_code {
None => {
let utc_schedule_time =
time::OffsetDateTime::now_utc() + time::Duration::minutes(1);
match token_with_none_error_code {
Some(token_with_retry_info) => {
let token_details = &token_with_retry_info
.token_status
.payment_processor_token_details;
let schedule_time = time::PrimitiveDateTime::new(
utc_schedule_time.date(),
utc_schedule_time.time(),
);
tokens_with_schedule_time = vec![ScheduledToken {
token_details: token_details.clone(),
schedule_time,
}];
tracing::debug!(
"Found payment processor token with no error code scheduling it for {schedule_time}",
);
break;
}
Some(_) => {
let utc_schedule_time = time::OffsetDateTime::now_utc() + time::Duration::minutes(1);
let schedule_time =
time::PrimitiveDateTime::new(utc_schedule_time.date(), utc_schedule_time.time());
tokens_with_schedule_time = vec![ScheduledToken {
token_details: token_details.clone(),
schedule_time,
}];
tracing::debug!(
"Found payment processor token with no error code, scheduling it for {schedule_time}",
);
}
None => {
for token_with_retry_info in processor_tokens.values() {
process_token_for_retry(state, token_with_retry_info, payment_intent)
.await?
.map(|token_with_schedule_time| {
@ -757,13 +889,27 @@ pub async fn call_decider_for_payment_processor_tokens_select_closet_time(
.min_by_key(|token| token.schedule_time)
.cloned();
let mut payment_processor_token_response;
match best_token {
None => {
// No tokens available for scheduling, unlock the connector customer status
// Check if all tokens are hard declined
let hard_decline_status = processor_tokens
.values()
.all(|token| token.token_status.is_hard_decline.unwrap_or(false));
RedisTokenManager::unlock_connector_customer_status(state, connector_customer_id)
.await
.change_context(errors::ProcessTrackerError::EApiErrorResponse)?;
tracing::debug!("No payment processor tokens available for scheduling");
Ok(None)
if hard_decline_status {
payment_processor_token_response = PaymentProcessorTokenResponse::HardDecline;
} else {
payment_processor_token_response = PaymentProcessorTokenResponse::None;
}
}
Some(token) => {
@ -778,9 +924,12 @@ pub async fn call_decider_for_payment_processor_tokens_select_closet_time(
.await
.change_context(errors::ProcessTrackerError::EApiErrorResponse)?;
Ok(Some(token.schedule_time))
payment_processor_token_response = PaymentProcessorTokenResponse::ScheduledTime {
scheduled_time: token.schedule_time,
};
}
}
Ok(payment_processor_token_response)
}
#[cfg(feature = "v2")]