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

feat: Kv changes for V2 feature (#8198)

Browse Source 
Co-authored-by: Akshay S <akshay.s@juspay.in>
Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>
This commit is contained in:
akshay-97
2025-06-26 19:15:22 +05:30
committed by GitHub
parent fb58cc9c2f
commit d2740f0322
15 changed files with 868 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
crates/storage_impl/src/customers.rs
View File

@ -23,6 +23,32 @@ use crate::{
impl KvStorePartition for customers::Customer {}
#[cfg(feature = "v2")]
mod label {
use common_utils::id_type;
pub(super) const MODEL_NAME: &str = "customer_v2";
pub(super) const CLUSTER_LABEL: &str = "cust";
pub(super) fn get_global_id_label(global_customer_id: &id_type::GlobalCustomerId) -> String {
format!(
"customer_global_id_{}",
global_customer_id.get_string_repr()
)
}
pub(super) fn get_merchant_scoped_id_label(
merchant_id: &id_type::MerchantId,
merchant_reference_id: &id_type::CustomerId,
) -> String {
format!(
"customer_mid_{}_mrefid_{}",
merchant_id.get_string_repr(),
merchant_reference_id.get_string_repr()
)
}
}
#[async_trait::async_trait]
impl<T: DatabaseStore> domain::CustomerInterface for kv_router_store::KVRouterStore<T> {
type Error = StorageError;
@ -283,22 +309,32 @@ impl<T: DatabaseStore> domain::CustomerInterface for kv_router_store::KVRouterSt
.construct_new()
.await
.change_context(StorageError::EncryptionError)?;
let storage_scheme = Box::pin(decide_storage_scheme::<_, customers::Customer>(
let decided_storage_scheme = Box::pin(decide_storage_scheme::<_, customers::Customer>(
self,
storage_scheme,
Op::Insert,
))
.await;
new_customer.update_storage_scheme(storage_scheme);
new_customer.update_storage_scheme(decided_storage_scheme);
let mut reverse_lookups = Vec::new();
if let Some(ref merchant_ref_id) = new_customer.merchant_reference_id {
let reverse_lookup_merchant_scoped_id =
label::get_merchant_scoped_id_label(&new_customer.merchant_id, merchant_ref_id);
reverse_lookups.push(reverse_lookup_merchant_scoped_id);
}
self.insert_resource(
state,
key_store,
storage_scheme,
decided_storage_scheme,
new_customer.clone().insert(&conn),
new_customer.clone().into(),
kv_router_store::InsertResourceParams {
insertable: kv::Insertable::Customer(new_customer.clone()),
reverse_lookups: vec![],
reverse_lookups,
identifier,
key,
resource_type: "customer",

10
crates/storage_impl/src/lib.rs
View File

@ -378,6 +378,16 @@ impl UniqueConstraints for diesel_models::PaymentAttempt {
}
}
#[cfg(feature = "v2")]
impl UniqueConstraints for diesel_models::PaymentAttempt {
fn unique_constraints(&self) -> Vec<String> {
vec![format!("pa_{}", self.id.get_string_repr())]
}
fn table_name(&self) -> &str {
"PaymentAttempt"
}
}
#[cfg(feature = "v1")]
impl UniqueConstraints for diesel_models::Refund {
fn unique_constraints(&self) -> Vec<String> {

326
crates/storage_impl/src/payments/payment_attempt.rs
View File

@ -6,16 +6,17 @@ use common_utils::{
fallback_reverse_lookup_not_found,
types::{ConnectorTransactionId, ConnectorTransactionIdTrait, CreatedBy},
};
#[cfg(feature = "v1")]
use diesel_models::payment_attempt::PaymentAttemptNew as DieselPaymentAttemptNew;
use diesel_models::{
enums::{
MandateAmountData as DieselMandateAmountData, MandateDataType as DieselMandateType,
MandateDetails as DieselMandateDetails, MerchantStorageScheme,
},
kv,
payment_attempt::PaymentAttempt as DieselPaymentAttempt,
reverse_lookup::{ReverseLookup, ReverseLookupNew},
};
#[cfg(feature = "v1")]
use diesel_models::{kv, payment_attempt::PaymentAttemptNew as DieselPaymentAttemptNew};
use error_stack::ResultExt;
#[cfg(feature = "v1")]
use hyperswitch_domain_models::payments::payment_attempt::PaymentAttemptNew;
@ -32,22 +33,21 @@ use hyperswitch_domain_models::{
use hyperswitch_domain_models::{
payments::payment_attempt::PaymentListFilters, payments::PaymentIntent,
};
#[cfg(feature = "v1")]
#[cfg(feature = "v2")]
use label::*;
use redis_interface::HsetnxReply;
use router_env::{instrument, tracing};
#[cfg(feature = "v2")]
use crate::kv_router_store::{FilterResourceParams, FindResourceBy, UpdateResourceParams};
use crate::{
diesel_error_to_data_error, errors,
errors::RedisErrorExt,
kv_router_store::KVRouterStore,
lookup::ReverseLookupInterface,
utils::{pg_connection_read, pg_connection_write},
DataModelExt, DatabaseStore, RouterStore,
};
#[cfg(feature = "v1")]
use crate::{
errors::RedisErrorExt,
redis::kv_store::{decide_storage_scheme, kv_wrapper, KvOperation, Op, PartitionKey},
utils::try_redis_get_else_try_database_get,
utils::{pg_connection_read, pg_connection_write, try_redis_get_else_try_database_get},
DataModelExt, DatabaseStore, RouterStore,
};
#[async_trait::async_trait]
@ -747,15 +747,98 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
payment_attempt: PaymentAttempt,
storage_scheme: MerchantStorageScheme,
) -> error_stack::Result<PaymentAttempt, errors::StorageError> {
// Ignoring storage scheme for v2 implementation
self.router_store
.insert_payment_attempt(
key_manager_state,
merchant_key_store,
payment_attempt,
storage_scheme,
)
.await
let decided_storage_scheme = Box::pin(decide_storage_scheme::<_, DieselPaymentAttempt>(
self,
storage_scheme,
Op::Insert,
))
.await;
match decided_storage_scheme {
MerchantStorageScheme::PostgresOnly => {
self.router_store
.insert_payment_attempt(
key_manager_state,
merchant_key_store,
payment_attempt,
decided_storage_scheme,
)
.await
}
MerchantStorageScheme::RedisKv => {
let key = PartitionKey::GlobalPaymentId {
id: &payment_attempt.payment_id,
};
let key_str = key.to_string();
let field = format!(
"{}_{}",
label::CLUSTER_LABEL,
payment_attempt.id.get_string_repr()
);
let diesel_payment_attempt_new = payment_attempt
.clone()
.construct_new()
.await
.change_context(errors::StorageError::EncryptionError)?;
let diesel_payment_attempt_for_redis: DieselPaymentAttempt =
Conversion::convert(payment_attempt.clone())
.await
.change_context(errors::StorageError::EncryptionError)?;
let redis_entry = kv::TypedSql {
op: kv::DBOperation::Insert {
insertable: Box::new(kv::Insertable::PaymentAttempt(Box::new(
diesel_payment_attempt_new.clone(),
))),
},
};
let reverse_lookup_attempt_id = ReverseLookupNew {
lookup_id: label::get_global_id_label(&payment_attempt.id),
pk_id: key_str.clone(),
sk_id: field.clone(),
source: "payment_attempt".to_string(),
updated_by: decided_storage_scheme.to_string(),
};
self.insert_reverse_lookup(reverse_lookup_attempt_id, decided_storage_scheme)
.await?;
if let Some(ref conn_txn_id_val) = payment_attempt.connector_payment_id {
let reverse_lookup_conn_txn_id = ReverseLookupNew {
lookup_id: label::get_profile_id_connector_transaction_label(
payment_attempt.profile_id.get_string_repr(),
conn_txn_id_val,
),
pk_id: key_str.clone(),
sk_id: field.clone(),
source: "payment_attempt".to_string(),
updated_by: decided_storage_scheme.to_string(),
};
self.insert_reverse_lookup(reverse_lookup_conn_txn_id, decided_storage_scheme)
.await?;
}
match Box::pin(kv_wrapper::<DieselPaymentAttempt, _, _>(
self,
KvOperation::HSetNx(&field, &diesel_payment_attempt_for_redis, redis_entry),
key,
))
.await
.map_err(|err| err.to_redis_failed_response(&key_str))?
.try_into_hsetnx()
{
Ok(HsetnxReply::KeyNotSet) => Err(errors::StorageError::DuplicateValue {
entity: "payment_attempt",
key: Some(payment_attempt.id.get_string_repr().to_owned()),
}
.into()),
Ok(HsetnxReply::KeySet) => Ok(payment_attempt),
Err(error) => Err(error.change_context(errors::StorageError::KVError)),
}
}
}
}
#[cfg(feature = "v1")]
@ -889,19 +972,48 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
key_manager_state: &KeyManagerState,
merchant_key_store: &MerchantKeyStore,
this: PaymentAttempt,
payment_attempt: PaymentAttemptUpdate,
payment_attempt_update: PaymentAttemptUpdate,
storage_scheme: MerchantStorageScheme,
) -> error_stack::Result<PaymentAttempt, errors::StorageError> {
// Ignoring storage scheme for v2 implementation
self.router_store
.update_payment_attempt(
key_manager_state,
merchant_key_store,
this,
payment_attempt,
storage_scheme,
)
let payment_attempt = Conversion::convert(this.clone())
.await
.change_context(errors::StorageError::DecryptionError)?;
let key = PartitionKey::GlobalPaymentId {
id: &this.payment_id,
};
let field = format!("{}_{}", label::CLUSTER_LABEL, this.id.get_string_repr());
let conn = pg_connection_write(self).await?;
let payment_attempt_internal =
diesel_models::PaymentAttemptUpdateInternal::from(payment_attempt_update);
let updated_payment_attempt = payment_attempt_internal
.clone()
.apply_changeset(payment_attempt.clone());
let updated_by = updated_payment_attempt.updated_by.to_owned();
let updated_payment_attempt_with_id = payment_attempt
.clone()
.update_with_attempt_id(&conn, payment_attempt_internal.clone());
Box::pin(self.update_resource(
key_manager_state,
merchant_key_store,
storage_scheme,
updated_payment_attempt_with_id,
updated_payment_attempt,
UpdateResourceParams {
updateable: kv::Updateable::PaymentAttemptUpdate(Box::new(
kv::PaymentAttemptUpdateMems {
orig: payment_attempt,
update_data: payment_attempt_internal,
},
)),
operation: Op::Update(key.clone(), &field, Some(updated_by.as_str())),
},
))
.await
}
#[cfg(feature = "v1")]
@ -1095,15 +1207,69 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
payment_id: &common_utils::id_type::GlobalPaymentId,
storage_scheme: MerchantStorageScheme,
) -> error_stack::Result<PaymentAttempt, errors::StorageError> {
// Ignoring storage scheme for v2 implementation
self.router_store
.find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
key_manager_state,
merchant_key_store,
payment_id,
storage_scheme,
)
.await
let database_call = || {
self.router_store
.find_payment_attempt_last_successful_or_partially_captured_attempt_by_payment_id(
key_manager_state,
merchant_key_store,
payment_id,
storage_scheme,
)
};
let decided_storage_scheme = Box::pin(decide_storage_scheme::<_, DieselPaymentAttempt>(
self,
storage_scheme,
Op::Find,
))
.await;
match decided_storage_scheme {
MerchantStorageScheme::PostgresOnly => database_call().await,
MerchantStorageScheme::RedisKv => {
let key = PartitionKey::GlobalPaymentId { id: payment_id };
let redis_fut = async {
let kv_result = kv_wrapper::<DieselPaymentAttempt, _, _>(
self,
KvOperation::<DieselPaymentAttempt>::Scan("pa_*"),
key.clone(),
)
.await?
.try_into_scan();
let payment_attempt = kv_result.and_then(|mut payment_attempts| {
payment_attempts.sort_by(|a, b| b.modified_at.cmp(&a.modified_at));
payment_attempts
.iter()
.find(|&pa| {
pa.status == diesel_models::enums::AttemptStatus::Charged
|| pa.status
== diesel_models::enums::AttemptStatus::PartialCharged
})
.cloned()
.ok_or(error_stack::report!(
redis_interface::errors::RedisError::NotFound
))
})?;
let merchant_id = payment_attempt.merchant_id.clone();
PaymentAttempt::convert_back(
key_manager_state,
payment_attempt,
merchant_key_store.key.get_inner(),
merchant_id.into(),
)
.await
.change_context(redis_interface::errors::RedisError::UnknownResult)
};
Box::pin(try_redis_get_else_try_database_get(
redis_fut,
database_call,
))
.await
}
}
}
#[cfg(feature = "v2")]
@ -1115,16 +1281,22 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
connector_transaction_id: &str,
storage_scheme: MerchantStorageScheme,
) -> CustomResult<PaymentAttempt, errors::StorageError> {
// Ignoring storage scheme for v2 implementation
self.router_store
.find_payment_attempt_by_profile_id_connector_transaction_id(
key_manager_state,
merchant_key_store,
let conn = pg_connection_read(self).await?;
self.find_resource_by_id(
key_manager_state,
merchant_key_store,
storage_scheme,
DieselPaymentAttempt::find_by_profile_id_connector_transaction_id(
&conn,
profile_id,
connector_transaction_id,
storage_scheme,
)
.await
),
FindResourceBy::LookupId(label::get_profile_id_connector_transaction_label(
profile_id.get_string_repr(),
connector_transaction_id,
)),
)
.await
}
#[instrument(skip_all)]
@ -1329,15 +1501,15 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
attempt_id: &common_utils::id_type::GlobalAttemptId,
storage_scheme: MerchantStorageScheme,
) -> error_stack::Result<PaymentAttempt, errors::StorageError> {
// Ignoring storage scheme for v2 implementation
self.router_store
.find_payment_attempt_by_id(
key_manager_state,
merchant_key_store,
attempt_id,
storage_scheme,
)
.await
let conn = pg_connection_read(self).await?;
self.find_resource_by_id(
key_manager_state,
merchant_key_store,
storage_scheme,
DieselPaymentAttempt::find_by_id(&conn, attempt_id),
FindResourceBy::LookupId(label::get_global_id_label(attempt_id)),
)
.await
}
#[cfg(feature = "v2")]
@ -1349,14 +1521,20 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
merchant_key_store: &MerchantKeyStore,
storage_scheme: MerchantStorageScheme,
) -> error_stack::Result<Vec<PaymentAttempt>, errors::StorageError> {
self.router_store
.find_payment_attempts_by_payment_intent_id(
key_manager_state,
payment_id,
merchant_key_store,
storage_scheme,
)
.await
let conn = pg_connection_read(self).await?;
self.filter_resources(
key_manager_state,
merchant_key_store,
storage_scheme,
DieselPaymentAttempt::find_by_payment_id(&conn, payment_id),
|_| true,
FilterResourceParams {
key: PartitionKey::GlobalPaymentId { id: payment_id },
pattern: "pa_*",
limit: None,
},
)
.await
}
#[cfg(feature = "v1")]
@ -2036,3 +2214,25 @@ async fn add_preprocessing_id_to_reverse_lookup<T: DatabaseStore>(
.insert_reverse_lookup(reverse_lookup_new, storage_scheme)
.await
}
#[cfg(feature = "v2")]
mod label {
pub(super) const MODEL_NAME: &str = "payment_attempt_v2";
pub(super) const CLUSTER_LABEL: &str = "pa";
pub(super) fn get_profile_id_connector_transaction_label(
profile_id: &str,
connector_transaction_id: &str,
) -> String {
format!(
"profile_{}_conn_txn_{}",
profile_id, connector_transaction_id
)
}
pub(super) fn get_global_id_label(
attempt_id: &common_utils::id_type::GlobalAttemptId,
) -> String {
format!("attempt_global_id_{}", attempt_id.get_string_repr())
}
}

274
crates/storage_impl/src/payments/payment_intent.rs
View File

@ -2,13 +2,22 @@
use api_models::payments::{AmountFilter, Order, SortBy, SortOn};
#[cfg(feature = "olap")]
use async_bb8_diesel::{AsyncConnection, AsyncRunQueryDsl};
#[cfg(feature = "v1")]
use common_utils::ext_traits::Encode;
use common_utils::{ext_traits::AsyncExt, types::keymanager::KeyManagerState};
#[cfg(feature = "v2")]
use common_utils::fallback_reverse_lookup_not_found;
use common_utils::{
ext_traits::{AsyncExt, Encode},
types::keymanager::KeyManagerState,
};
#[cfg(feature = "olap")]
use diesel::{associations::HasTable, ExpressionMethods, JoinOnDsl, QueryDsl};
#[cfg(feature = "v1")]
use diesel_models::payment_intent::PaymentIntentUpdate as DieselPaymentIntentUpdate;
#[cfg(feature = "v2")]
use diesel_models::payment_intent::PaymentIntentUpdateInternal;
#[cfg(feature = "olap")]
use diesel_models::query::generics::db_metrics;
#[cfg(feature = "v2")]
use diesel_models::reverse_lookup::ReverseLookupNew;
#[cfg(all(feature = "v1", feature = "olap"))]
use diesel_models::schema::{
payment_attempt::{self as payment_attempt_schema, dsl as pa_dsl},
@ -20,10 +29,8 @@ use diesel_models::schema_v2::{
payment_intent::dsl as pi_dsl,
};
use diesel_models::{
enums::MerchantStorageScheme, payment_intent::PaymentIntent as DieselPaymentIntent,
enums::MerchantStorageScheme, kv, payment_intent::PaymentIntent as DieselPaymentIntent,
};
#[cfg(feature = "v1")]
use diesel_models::{kv, payment_intent::PaymentIntentUpdate as DieselPaymentIntentUpdate};
use error_stack::ResultExt;
#[cfg(feature = "olap")]
use hyperswitch_domain_models::payments::{
@ -37,7 +44,6 @@ use hyperswitch_domain_models::{
PaymentIntent,
},
};
#[cfg(feature = "v1")]
use redis_interface::HsetnxReply;
#[cfg(feature = "olap")]
use router_env::logger;
@ -47,17 +53,14 @@ use router_env::{instrument, tracing};
use crate::connection;
use crate::{
diesel_error_to_data_error,
errors::StorageError,
errors::{RedisErrorExt, StorageError},
kv_router_store::KVRouterStore,
utils::{pg_connection_read, pg_connection_write},
redis::kv_store::{decide_storage_scheme, kv_wrapper, KvOperation, Op, PartitionKey},
utils::{self, pg_connection_read, pg_connection_write},
DatabaseStore,
};
#[cfg(feature = "v1")]
use crate::{
errors::RedisErrorExt,
redis::kv_store::{decide_storage_scheme, kv_wrapper, KvOperation, Op, PartitionKey},
utils,
};
#[cfg(feature = "v2")]
use crate::{errors, lookup::ReverseLookupInterface};
#[async_trait::async_trait]
impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
@ -163,7 +166,68 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
}
MerchantStorageScheme::RedisKv => {
todo!("Implement payment intent insert for kv")
let id = payment_intent.id.clone();
let key = PartitionKey::GlobalPaymentId { id: &id };
let field = format!("pi_{}", id.get_string_repr());
let key_str = key.to_string();
let new_payment_intent = payment_intent
.clone()
.construct_new()
.await
.change_context(StorageError::EncryptionError)?;
let redis_entry = kv::TypedSql {
op: kv::DBOperation::Insert {
insertable: Box::new(kv::Insertable::PaymentIntent(Box::new(
new_payment_intent,
))),
},
};
let diesel_payment_intent = payment_intent
.clone()
.convert()
.await
.change_context(StorageError::EncryptionError)?;
if let Some(merchant_reference_id) = &payment_intent.merchant_reference_id {
let reverse_lookup = ReverseLookupNew {
lookup_id: format!(
"pi_merchant_reference_{}_{}",
payment_intent.profile_id.get_string_repr(),
merchant_reference_id.get_string_repr()
),
pk_id: key_str.clone(),
sk_id: field.clone(),
source: "payment_intent".to_string(),
updated_by: storage_scheme.to_string(),
};
self.insert_reverse_lookup(reverse_lookup, storage_scheme)
.await?;
}
match Box::pin(kv_wrapper::<DieselPaymentIntent, _, _>(
self,
KvOperation::<DieselPaymentIntent>::HSetNx(
&field,
&diesel_payment_intent,
redis_entry,
),
key,
))
.await
.map_err(|err| err.to_redis_failed_response(&key_str))?
.try_into_hsetnx()
{
Ok(HsetnxReply::KeyNotSet) => Err(StorageError::DuplicateValue {
entity: "payment_intent",
key: Some(key_str),
}
.into()),
Ok(HsetnxReply::KeySet) => Ok(payment_intent),
Err(error) => Err(error.change_context(StorageError::KVError)),
}
}
}
}
@ -300,7 +364,59 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
.await
}
MerchantStorageScheme::RedisKv => {
todo!()
let id = this.id.clone();
let merchant_id = this.merchant_id.clone();
let key = PartitionKey::GlobalPaymentId { id: &id };
let field = format!("pi_{}", id.get_string_repr());
let key_str = key.to_string();
let diesel_intent_update =
PaymentIntentUpdateInternal::try_from(payment_intent_update)
.change_context(StorageError::DeserializationFailed)?;
let origin_diesel_intent = this
.convert()
.await
.change_context(StorageError::EncryptionError)?;
let diesel_intent = diesel_intent_update
.clone()
.apply_changeset(origin_diesel_intent.clone());
let redis_value = diesel_intent
.encode_to_string_of_json()
.change_context(StorageError::SerializationFailed)?;
let redis_entry = kv::TypedSql {
op: kv::DBOperation::Update {
updatable: Box::new(kv::Updateable::PaymentIntentUpdate(Box::new(
kv::PaymentIntentUpdateMems {
orig: origin_diesel_intent,
update_data: diesel_intent_update,
},
))),
},
};
Box::pin(kv_wrapper::<(), _, _>(
self,
KvOperation::<DieselPaymentIntent>::Hset((&field, redis_value), redis_entry),
key,
))
.await
.map_err(|err| err.to_redis_failed_response(&key_str))?
.try_into_hset()
.change_context(StorageError::KVError)?;
let payment_intent = PaymentIntent::convert_back(
state,
diesel_intent,
merchant_key_store.key.get_inner(),
merchant_id.into(),
)
.await
.change_context(StorageError::DecryptionError)?;
Ok(payment_intent)
}
}
}
@ -372,18 +488,50 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
state: &KeyManagerState,
id: &common_utils::id_type::GlobalPaymentId,
merchant_key_store: &MerchantKeyStore,
_storage_scheme: MerchantStorageScheme,
storage_scheme: MerchantStorageScheme,
) -> error_stack::Result<PaymentIntent, StorageError> {
let conn: bb8::PooledConnection<
'_,
async_bb8_diesel::ConnectionManager<diesel::PgConnection>,
> = pg_connection_read(self).await?;
let diesel_payment_intent = DieselPaymentIntent::find_by_global_id(&conn, id)
.await
.map_err(|er| {
let new_err = diesel_error_to_data_error(*er.current_context());
er.change_context(new_err)
})?;
let storage_scheme = Box::pin(decide_storage_scheme::<_, DieselPaymentIntent>(
self,
storage_scheme,
Op::Find,
))
.await;
let database_call = || async {
let conn: bb8::PooledConnection<
'_,
async_bb8_diesel::ConnectionManager<diesel::PgConnection>,
> = pg_connection_read(self).await?;
DieselPaymentIntent::find_by_global_id(&conn, id)
.await
.map_err(|er| {
let new_err = diesel_error_to_data_error(*er.current_context());
er.change_context(new_err)
})
};
let diesel_payment_intent = match storage_scheme {
MerchantStorageScheme::PostgresOnly => database_call().await,
MerchantStorageScheme::RedisKv => {
let key = PartitionKey::GlobalPaymentId { id };
let field = format!("pi_{}", id.get_string_repr());
Box::pin(utils::try_redis_get_else_try_database_get(
async {
Box::pin(kv_wrapper::<DieselPaymentIntent, _, _>(
self,
KvOperation::<DieselPaymentIntent>::HGet(&field),
key,
))
.await?
.try_into_hget()
},
database_call,
))
.await
}
}?;
let merchant_id = diesel_payment_intent.merchant_id.clone();
@ -391,7 +539,7 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
state,
diesel_payment_intent,
merchant_key_store.key.get_inner(),
merchant_id.to_owned().into(),
merchant_id.into(),
)
.await
.change_context(StorageError::DecryptionError)
@ -523,7 +671,68 @@ impl<T: DatabaseStore> PaymentIntentInterface for KVRouterStore<T> {
.await
}
MerchantStorageScheme::RedisKv => {
todo!()
let lookup_id = format!(
"pi_merchant_reference_{}_{}",
profile_id.get_string_repr(),
merchant_reference_id.get_string_repr()
);
let lookup = fallback_reverse_lookup_not_found!(
self.get_lookup_by_lookup_id(&lookup_id, *storage_scheme)
.await,
self.router_store
.find_payment_intent_by_merchant_reference_id_profile_id(
state,
merchant_reference_id,
profile_id,
merchant_key_store,
storage_scheme,
)
.await
);
let key = PartitionKey::CombinationKey {
combination: &lookup.pk_id,
};
let database_call = || async {
let conn = pg_connection_read(self).await?;
DieselPaymentIntent::find_by_merchant_reference_id_profile_id(
&conn,
merchant_reference_id,
profile_id,
)
.await
.map_err(|er| {
let new_err = diesel_error_to_data_error(*er.current_context());
er.change_context(new_err)
})
};
let diesel_payment_intent = Box::pin(utils::try_redis_get_else_try_database_get(
async {
Box::pin(kv_wrapper::<DieselPaymentIntent, _, _>(
self,
KvOperation::<DieselPaymentIntent>::HGet(&lookup.sk_id),
key,
))
.await?
.try_into_hget()
},
database_call,
))
.await?;
let merchant_id = diesel_payment_intent.merchant_id.clone();
PaymentIntent::convert_back(
state,
diesel_payment_intent,
merchant_key_store.key.get_inner(),
merchant_id.into(),
)
.await
.change_context(StorageError::DecryptionError)
}
}
}
@ -607,9 +816,8 @@ impl<T: DatabaseStore> PaymentIntentInterface for crate::RouterStore<T> {
_storage_scheme: MerchantStorageScheme,
) -> error_stack::Result<PaymentIntent, StorageError> {
let conn = pg_connection_write(self).await?;
let diesel_payment_intent_update =
diesel_models::PaymentIntentUpdateInternal::try_from(payment_intent)
.change_context(StorageError::DeserializationFailed)?;
let diesel_payment_intent_update = PaymentIntentUpdateInternal::try_from(payment_intent)
.change_context(StorageError::DeserializationFailed)?;
let diesel_payment_intent = this
.convert()
.await

10
crates/storage_impl/src/redis/kv_store.rs
View File

@ -55,6 +55,10 @@ pub enum PartitionKey<'a> {
GlobalId {
id: &'a str,
},
#[cfg(feature = "v2")]
GlobalPaymentId {
id: &'a common_utils::id_type::GlobalPaymentId,
},
}
// PartitionKey::MerchantIdPaymentId {merchant_id, payment_id}
impl std::fmt::Display for PartitionKey<'_> {
@ -108,7 +112,11 @@ impl std::fmt::Display for PartitionKey<'_> {
)),
#[cfg(feature = "v2")]
PartitionKey::GlobalId { id } => f.write_str(&format!("cust_{id}",)),
PartitionKey::GlobalId { id } => f.write_str(&format!("global_cust_{id}",)),
#[cfg(feature = "v2")]
PartitionKey::GlobalPaymentId { id } => {
f.write_str(&format!("global_payment_{}", id.get_string_repr()))
}
}
}
}