feat: add an api for toggle KV for all merchants

crates/api_models/src/admin.rs

@@ -826,6 +826,22 @@ 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/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/router/src/core/admin.rs

@@ -1419,6 +1419,36 @@ pub async fn kv_for_merchant(
  ))
 }
 
+pub async fn toggle_kv_for_all_merchants(
+ state: AppState,
+ enable: bool,
+) -> RouterResponse<api_models::admin::ToggleAllKVResponse> {
+ let db = state.store.as_ref();
+ let storage_scheme = if enable {
+ MerchantStorageScheme::RedisKv
+ } else {
+ MerchantStorageScheme::PostgresOnly
+ };
+
+ let total_update = db
+ .update_all_merchant_account(storage::MerchantAccountUpdate::StorageSchemeUpdate {
+ storage_scheme,
+ })
+ .await
+ .map_err(|error| {
+ error
+ .change_context(errors::ApiErrorResponse::InternalServerError)
+ .attach_printable("Failed to switch merchant_storage_scheme for all merchants")
+ })?;
+
+ Ok(service_api::ApplicationResponse::Json(
+ api_models::admin::ToggleAllKVResponse {
+ total_updated: total_update,
+ kv_enabled: enable,
+ },
+ ))
+}
+
 pub async fn check_merchant_account_kv_status(
  state: AppState,
  merchant_id: String,

crates/router/src/db/kafka_store.rs

@@ -817,6 +817,15 @@ impl MerchantAccountInterface for KafkaStore {
  .await
  }
 
+ async fn update_all_merchant_account(
+ &self,
+ merchant_account: storage::MerchantAccountUpdate,
+ ) -> CustomResult<usize, errors::StorageError> {
+ self.diesel_store
+ .update_all_merchant_account(merchant_account)
+ .await
+ }
+
  async fn find_merchant_account_by_publishable_key(
  &self,
  publishable_key: &str,

crates/router/src/db/merchant_account.rs

@@ -2,6 +2,7 @@
 use std::collections::HashMap;
 
 use common_utils::ext_traits::AsyncExt;
+use diesel_models::MerchantAccountUpdateInternal;
 use error_stack::{report, ResultExt};
 use router_env::{instrument, tracing};
 #[cfg(feature = "accounts_cache")]
@@ -40,6 +41,11 @@ where
  merchant_key_store: &domain::MerchantKeyStore,
  ) -> CustomResult<domain::MerchantAccount, errors::StorageError>;
 
+ async fn update_all_merchant_account(
+ &self,
+ merchant_account: storage::MerchantAccountUpdate,
+ ) -> CustomResult<usize, errors::StorageError>;
+
  async fn update_merchant(
  &self,
  this: domain::MerchantAccount,
@@ -354,6 +360,38 @@ impl MerchantAccountInterface for Store {
 
  Ok(merchant_accounts)
  }
+
+ async fn update_all_merchant_account(
+ &self,
+ merchant_account: storage::MerchantAccountUpdate,
+ ) -> CustomResult<usize, errors::StorageError> {
+ let conn = connection::pg_connection_read(self).await?;
+
+ let db_func = || async {
+ storage::MerchantAccount::update_all_merchant_accounts(
+ &conn,
+ MerchantAccountUpdateInternal::from(merchant_account),
+ )
+ .await
+ .map_err(|error| report!(errors::StorageError::from(error)))
+ };
+
+ let total;
+ #[cfg(not(feature = "accounts_cache"))]
+ {
+ let ma = db_func().await?;
+ total = ma.len();
+ }
+
+ #[cfg(feature = "accounts_cache")]
+ {
+ let ma = db_func().await?;
+ publish_and_redact_all_merchant_account_cache(self, &ma).await?;
+ total = ma.len();
+ }
+
+ Ok(total)
+ }
 }
 
 #[async_trait::async_trait]
@@ -433,6 +471,13 @@ impl MerchantAccountInterface for MockDb {
  Err(errors::StorageError::MockDbError)?
  }
 
+ async fn update_all_merchant_account(
+ &self,
+ _merchant_account_update: storage::MerchantAccountUpdate,
+ ) -> CustomResult<usize, errors::StorageError> {
+ Err(errors::StorageError::MockDbError)?
+ }
+
  async fn delete_merchant_account_by_merchant_id(
  &self,
  _merchant_id: &str,
@@ -477,3 +522,22 @@ async fn publish_and_redact_merchant_account_cache(
  super::cache::publish_into_redact_channel(store, cache_keys).await?;
  Ok(())
 }
+
+#[cfg(feature = "accounts_cache")]
+async fn publish_and_redact_all_merchant_account_cache(
+ store: &dyn super::StorageInterface,
+ merchant_accounts: &Vec<storage::MerchantAccount>,
+) -> CustomResult<(), errors::StorageError> {
+ let merchant_ids = merchant_accounts.iter().map(|m| m.merchant_id.clone());
+ let publishable_keys = merchant_accounts
+ .iter()
+ .filter_map(|m| m.publishable_key.clone());
+
+ let cache_keys: Vec<CacheKind<'_>> = merchant_ids
+ .chain(publishable_keys)
+ .map(|s| CacheKind::Accounts(s.into()))
+ .collect();
+
+ super::cache::publish_into_redact_channel(store, cache_keys).await?;
+ Ok(())
+}

crates/router/src/routes/admin.rs

@@ -458,6 +458,31 @@ pub async fn merchant_account_toggle_kv(
  )
  .await
 }
+
+/// Merchant Account - Toggle KV
+///
+/// Toggle KV mode for all Merchant Accounts
+#[instrument(skip_all)]
+pub async fn merchant_account_toggle_all_kv(
+ state: web::Data<AppState>,
+ req: HttpRequest,
+ json_payload: web::Json<admin::ToggleAllKVRequest>,
+) -> HttpResponse {
+ let flow = Flow::ConfigKeyUpdate;
+ let payload = json_payload.into_inner();
+
+ api::server_wrap(
+ flow,
+ state,
+ &req,
+ payload,
+ |state, _, payload, _| toggle_kv_for_all_merchants(state, payload.kv_enabled),
+ &auth::AdminApiAuth,
+ api_locking::LockAction::NotApplicable,
+ )
+ .await
+}
+
 #[instrument(skip_all, fields(flow = ?Flow::BusinessProfileCreate))]
 pub async fn business_profile_create(
  state: web::Data<AppState>,

crates/router/src/routes/app.rs

@@ -864,6 +864,7 @@ impl MerchantAccount {
  .route(web::post().to(merchant_account_toggle_kv))
  .route(web::get().to(merchant_account_kv_status)),
  )
+ .service(web::resource("/kv").route(web::post().to(merchant_account_toggle_all_kv)))
  .service(
  web::resource("/{id}")
  .route(web::get().to(retrieve_merchant_account))

crates/router/src/types/api/admin.rs

@@ -3,7 +3,8 @@ pub use api_models::admin::{
  MerchantAccountDeleteResponse, MerchantAccountResponse, MerchantAccountUpdate,
  MerchantConnectorCreate, MerchantConnectorDeleteResponse, MerchantConnectorDetails,
  MerchantConnectorDetailsWrap, MerchantConnectorId, MerchantConnectorResponse, MerchantDetails,
- MerchantId, PaymentMethodsEnabled, ToggleKVRequest, ToggleKVResponse, WebhookDetails,
+ MerchantId, PaymentMethodsEnabled, ToggleAllKVRequest, ToggleAllKVResponse, ToggleKVRequest,
+ ToggleKVResponse, WebhookDetails,
 };
 use common_utils::ext_traits::{Encode, ValueExt};
 use error_stack::ResultExt;