Issue:
Bug: [FEAT] add an API to toggle KV for all the merchants





Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->
- [x] New feature

## Description
<!-- Describe your changes in detail -->
Add an API to toggle KV for all merchants

### Additional Changes
- [x] This PR modifies the API contract

## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
Add an API to toggle KV for all merchants

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
- Enable KV for all merchants
```bash
curl --location 'http://localhost:8080/accounts/kv' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "kv_enabled": true
}'
```
Check if any merchant_account has `postgres_only`. 

```sql
SELECT storage_scheme from merchant_account where storage_scheme='postgres_only'
```
It should return 0 rows

-Disable KV for all merchants
```bash
curl --location 'http://localhost:8080/accounts/kv' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "kv_enabled": false
}'
```
Check if any merchant_account has `redis_kv`
```sql
SELECT storage_scheme from merchant_account where storage_scheme='redis_kv'
```
It should return 0 rows

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code

Testing:

- Enable KV for all merchants
```bash
curl --location 'http://localhost:8080/accounts/kv' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "kv_enabled": true
}'
```
Check if any merchant_account has `postgres_only`. 

```sql
SELECT storage_scheme from merchant_account where storage_scheme='postgres_only'
```
It should return 0 rows

-Disable KV for all merchants
```bash
curl --location 'http://localhost:8080/accounts/kv' \
--header 'Content-Type: application/json' \
--header 'api-key: test_admin' \
--data '{
    "kv_enabled": false
}'
```
Check if any merchant_account has `redis_kv`
```sql
SELECT storage_scheme from merchant_account where storage_scheme='redis_kv'
```
It should return 0 rows

Patch:
diff --git a/crates/api_models/src/admin.rs b/crates/api_models/src/admin.rs
index 2e288140a9f..12d9832ada6 100644
--- a/crates/api_models/src/admin.rs
+++ b/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".
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index c9ae775fed8..9c26576e77b 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -60,6 +60,8 @@ impl_misc_api_event_type!(
     RevokeApiKeyResponse,
     ToggleKVResponse,
     ToggleKVRequest,
+    ToggleAllKVRequest,
+    ToggleAllKVResponse,
     MerchantAccountDeleteResponse,
     MerchantAccountUpdate,
     CardInfoResponse,
diff --git a/crates/diesel_models/src/query/merchant_account.rs b/crates/diesel_models/src/query/merchant_account.rs
index 1d4eef75206..dd2f284305b 100644
--- a/crates/diesel_models/src/query/merchant_account.rs
+++ b/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
+    }
 }
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index 288de66de07..af707087ce3 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -1420,6 +1420,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,
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index bfe0592fd3d..88579892d2f 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/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,
diff --git a/crates/router/src/db/merchant_account.rs b/crates/router/src/db/merchant_account.rs
index 08d0c279047..c323ca4abb1 100644
--- a/crates/router/src/db/merchant_account.rs
+++ b/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(())
+}
diff --git a/crates/router/src/routes/admin.rs b/crates/router/src/routes/admin.rs
index d9cadf002c1..2d1b77460a5 100644
--- a/crates/router/src/routes/admin.rs
+++ b/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>,
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 21cc994381e..f4eded24be6 100644
--- a/crates/router/src/routes/app.rs
+++ b/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))
diff --git a/crates/router/src/types/api/admin.rs b/crates/router/src/types/api/admin.rs
index fcbd97d5cb1..8fc3ebb721f 100644
--- a/crates/router/src/types/api/admin.rs
+++ b/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;