feat: Soft kill kv

config/development.toml

@@ -560,6 +560,7 @@ delay_between_retries_in_milliseconds = 500
 
 [kv_config]
 ttl = 900 # 15 * 60 seconds
+soft_kill = false
 
 [frm]
 enabled = true

config/docker_compose.toml

@@ -450,6 +450,7 @@ queue_strategy = "Fifo"
 
 [kv_config]
 ttl = 900 # 15 * 60 seconds
+soft_kill = false
 
 [frm]
 enabled = true

crates/api_models/src/admin.rs

@@ -826,6 +826,23 @@ pub struct ToggleKVRequest {
     pub kv_enabled: bool,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct ToggleAllKVRequest {
+    /// Status of KV for the specific merchant
+    #[schema(example = true)]
+    pub kv_enabled: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct ToggleAllKVResponse {
+    ///Total number of updated merchants
+    #[schema(example = 20)]
+    pub total_updated: usize,
+    /// Status of KV for the specific merchant
+    #[schema(example = true)]
+    pub kv_enabled: bool,
+}
+
 #[derive(Debug, Clone, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize, ToSchema)]
 pub struct MerchantConnectorDetailsWrap {
     /// Creds Identifier is to uniquely identify the credentials. Do not send any sensitive info in this field. And do not send the string "null".

crates/api_models/src/events.rs

@@ -60,6 +60,8 @@ impl_misc_api_event_type!(
     RevokeApiKeyResponse,
     ToggleKVResponse,
     ToggleKVRequest,
+    ToggleAllKVRequest,
+    ToggleAllKVResponse,
     MerchantAccountDeleteResponse,
     MerchantAccountUpdate,
     CardInfoResponse,

crates/diesel_models/src/customers.rs

@@ -1,3 +1,4 @@
+use common_enums::MerchantStorageScheme;
 use common_utils::pii;
 use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
 use time::PrimitiveDateTime;
@@ -21,6 +22,13 @@ pub struct CustomerNew {
     pub created_at: PrimitiveDateTime,
     pub modified_at: PrimitiveDateTime,
     pub address_id: Option<String>,
+    pub updated_by: Option<String>,
+}
+
+impl CustomerNew {
+    pub fn update_storage_scheme(&mut self, storage_scheme: MerchantStorageScheme) {
+        self.updated_by = Some(storage_scheme.to_string());
+    }
 }
 
 impl From<CustomerNew> for Customer {
@@ -40,6 +48,7 @@ impl From<CustomerNew> for Customer {
             modified_at: customer_new.modified_at,
             address_id: customer_new.address_id,
             default_payment_method_id: None,
+            updated_by: customer_new.updated_by,
         }
     }
 }
@@ -61,6 +70,7 @@ pub struct Customer {
     pub modified_at: PrimitiveDateTime,
     pub address_id: Option<String>,
     pub default_payment_method_id: Option<String>,
+    pub updated_by: Option<String>,
 }
 
 #[derive(
@@ -84,6 +94,7 @@ pub struct CustomerUpdateInternal {
     pub connector_customer: Option<serde_json::Value>,
     pub address_id: Option<String>,
     pub default_payment_method_id: Option<Option<String>>,
+    pub updated_by: Option<String>,
 }
 
 impl CustomerUpdateInternal {

crates/diesel_models/src/mandate.rs

@@ -1,3 +1,4 @@
+use common_enums::MerchantStorageScheme;
 use common_utils::pii;
 use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
 use masking::Secret;
@@ -32,6 +33,7 @@ pub struct Mandate {
     pub connector_mandate_ids: Option<pii::SecretSerdeValue>,
     pub original_payment_id: Option<String>,
     pub merchant_connector_id: Option<String>,
+    pub updated_by: Option<String>,
 }
 
 #[derive(
@@ -69,6 +71,13 @@ pub struct MandateNew {
     pub connector_mandate_ids: Option<pii::SecretSerdeValue>,
     pub original_payment_id: Option<String>,
     pub merchant_connector_id: Option<String>,
+    pub updated_by: Option<String>,
+}
+
+impl MandateNew {
+    pub fn update_storage_scheme(&mut self, storage_scheme: MerchantStorageScheme) {
+        self.updated_by = Some(storage_scheme.to_string());
+    }
 }
 
 #[derive(Debug)]
@@ -90,6 +99,17 @@ pub enum MandateUpdate {
     },
 }
 
+impl MandateUpdate {
+    pub fn convert_to_mandate_update(
+        self,
+        storage_scheme: MerchantStorageScheme,
+    ) -> MandateUpdateInternal {
+        let mut updated_object = MandateUpdateInternal::from(self);
+        updated_object.updated_by = Some(storage_scheme.to_string());
+        updated_object
+    }
+}
+
 #[derive(Clone, Eq, PartialEq, Copy, Debug, Default, serde::Serialize, serde::Deserialize)]
 pub struct SingleUseMandate {
     pub amount: i64,
@@ -113,6 +133,7 @@ pub struct MandateUpdateInternal {
     connector_mandate_id: Option<String>,
     payment_method_id: Option<String>,
     original_payment_id: Option<String>,
+    updated_by: Option<String>,
 }
 
 impl From<MandateUpdate> for MandateUpdateInternal {
@@ -125,6 +146,7 @@ impl From<MandateUpdate> for MandateUpdateInternal {
                 connector_mandate_id: None,
                 payment_method_id: None,
                 original_payment_id: None,
+                updated_by: None,
             },
             MandateUpdate::CaptureAmountUpdate { amount_captured } => Self {
                 mandate_status: None,
@@ -133,6 +155,7 @@ impl From<MandateUpdate> for MandateUpdateInternal {
                 connector_mandate_id: None,
                 payment_method_id: None,
                 original_payment_id: None,
+                updated_by: None,
             },
             MandateUpdate::ConnectorReferenceUpdate {
                 connector_mandate_ids,
@@ -165,6 +188,7 @@ impl MandateUpdateInternal {
             connector_mandate_id,
             payment_method_id,
             original_payment_id,
+            updated_by,
         } = self;
 
         Mandate {
@@ -174,6 +198,7 @@ impl MandateUpdateInternal {
             connector_mandate_id: connector_mandate_id.map_or(source.connector_mandate_id, Some),
             payment_method_id: payment_method_id.unwrap_or(source.payment_method_id),
             original_payment_id: original_payment_id.map_or(source.original_payment_id, Some),
+            updated_by: updated_by.map_or(source.updated_by, Some),
             ..source
         }
     }
@@ -208,6 +233,7 @@ impl From<&MandateNew> for Mandate {
             connector_mandate_ids: mandate_new.connector_mandate_ids.clone(),
             original_payment_id: mandate_new.original_payment_id.clone(),
             merchant_connector_id: mandate_new.merchant_connector_id.clone(),
+            updated_by: mandate_new.updated_by.clone(),
         }
     }
 }

crates/diesel_models/src/payment_method.rs

@@ -1,3 +1,4 @@
+use common_enums::MerchantStorageScheme;
 use common_utils::pii;
 use diesel::{AsChangeset, Identifiable, Insertable, Queryable};
 use masking::Secret;
@@ -41,6 +42,7 @@ pub struct PaymentMethod {
     pub network_transaction_id: Option<String>,
     pub client_secret: Option<String>,
     pub payment_method_billing_address: Option<Encryption>,
+    pub updated_by: Option<String>,
 }
 
 #[derive(
@@ -77,6 +79,13 @@ pub struct PaymentMethodNew {
     pub network_transaction_id: Option<String>,
     pub client_secret: Option<String>,
     pub payment_method_billing_address: Option<Encryption>,
+    pub updated_by: Option<String>,
+}
+
+impl PaymentMethodNew {
+    pub fn update_storage_scheme(&mut self, storage_scheme: MerchantStorageScheme) {
+        self.updated_by = Some(storage_scheme.to_string());
+    }
 }
 
 #[derive(Debug, Eq, PartialEq, Deserialize, Serialize)]
@@ -116,6 +125,17 @@ pub enum PaymentMethodUpdate {
     },
 }
 
+impl PaymentMethodUpdate {
+    pub fn convert_to_payment_method_update(
+        self,
+        storage_scheme: MerchantStorageScheme,
+    ) -> PaymentMethodUpdateInternal {
+        let mut update_internal: PaymentMethodUpdateInternal = self.into();
+        update_internal.updated_by = Some(storage_scheme.to_string());
+        update_internal
+    }
+}
+
 #[derive(
     Clone, Debug, Default, AsChangeset, router_derive::DebugAsDisplay, Serialize, Deserialize,
 )]
@@ -131,6 +151,7 @@ pub struct PaymentMethodUpdateInternal {
     connector_mandate_details: Option<serde_json::Value>,
     payment_method_type: Option<storage_enums::PaymentMethodType>,
     payment_method_issuer: Option<String>,
+    updated_by: Option<String>,
 }
 
 impl PaymentMethodUpdateInternal {
@@ -148,6 +169,7 @@ impl PaymentMethodUpdateInternal {
             network_transaction_id,
             status,
             connector_mandate_details,
+            updated_by,
             ..
         } = self;
 
@@ -160,6 +182,7 @@ impl PaymentMethodUpdateInternal {
             status: status.unwrap_or(source.status),
             connector_mandate_details: connector_mandate_details
                 .map_or(source.connector_mandate_details, Some),
+            updated_by: updated_by.map_or(source.updated_by, Some),
             ..source
         }
     }
@@ -179,6 +202,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 connector_mandate_details: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                updated_by: None,
             },
             PaymentMethodUpdate::PaymentMethodDataUpdate {
                 payment_method_data,
@@ -193,6 +217,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 connector_mandate_details: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                updated_by: None,
             },
             PaymentMethodUpdate::LastUsedUpdate { last_used_at } => Self {
                 metadata: None,
@@ -205,6 +230,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 connector_mandate_details: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                updated_by: None,
             },
             PaymentMethodUpdate::NetworkTransactionIdAndStatusUpdate {
                 network_transaction_id,
@@ -220,6 +246,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 connector_mandate_details: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                updated_by: None,
             },
             PaymentMethodUpdate::StatusUpdate { status } => Self {
                 metadata: None,
@@ -232,6 +259,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 connector_mandate_details: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                updated_by: None,
             },
             PaymentMethodUpdate::AdditionalDataUpdate {
                 payment_method_data,
@@ -251,6 +279,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 connector_mandate_details: None,
                 payment_method_issuer,
                 payment_method_type,
+                updated_by: None,
             },
             PaymentMethodUpdate::ConnectorMandateDetailsUpdate {
                 connector_mandate_details,
@@ -265,6 +294,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 network_transaction_id: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                updated_by: None,
             },
         }
     }
@@ -305,6 +335,7 @@ impl From<&PaymentMethodNew> for PaymentMethod {
             payment_method_billing_address: payment_method_new
                 .payment_method_billing_address
                 .clone(),
+            updated_by: payment_method_new.updated_by.clone(),
         }
     }
 }

crates/diesel_models/src/query/merchant_account.rs

@@ -123,4 +123,16 @@ impl MerchantAccount {
         )
         .await
     }
+
+    pub async fn update_all_merchant_accounts(
+        conn: &PgPooledConn,
+        merchant_account: MerchantAccountUpdateInternal,
+    ) -> StorageResult<Vec<Self>> {
+        generics::generic_update_with_results::<<Self as HasTable>::Table, _, _, _>(
+            conn,
+            dsl::merchant_id.ne_all(vec![""]),
+            merchant_account,
+        )
+        .await
+    }
 }

crates/diesel_models/src/schema.rs

@@ -295,6 +295,8 @@ diesel::table! {
         address_id -> Nullable<Varchar>,
         #[max_length = 64]
         default_payment_method_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        updated_by -> Nullable<Varchar>,
     }
 }
 
@@ -590,6 +592,8 @@ diesel::table! {
         original_payment_id -> Nullable<Varchar>,
         #[max_length = 32]
         merchant_connector_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        updated_by -> Nullable<Varchar>,
     }
 }
 
@@ -928,6 +932,8 @@ diesel::table! {
         #[max_length = 128]
         client_secret -> Nullable<Varchar>,
         payment_method_billing_address -> Nullable<Bytea>,
+        #[max_length = 64]
+        updated_by -> Nullable<Varchar>,
     }
 }
 

crates/router/src/configs/defaults.rs

@@ -123,7 +123,10 @@ impl Default for super::settings::DrainerSettings {
 #[cfg(feature = "kv_store")]
 impl Default for super::settings::KvConfig {
     fn default() -> Self {
-        Self { ttl: 900 }
+        Self {
+            ttl: 900,
+            soft_kill: Some(false),
+        }
     }
 }