feat: add an api for toggle KV for all merchants (#4600)

crates/api_models/src/admin.rs

@@ -829,6 +829,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

@@ -1427,6 +1427,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

@@ -828,6 +828,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: &[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

@@ -866,6 +866,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;