Bug: [FEATURE] Update Payment Connector Create to incorporate recipient creation



### Feature Description

In Open banking payments, some connector offer a recipient creation flow, where a `recipient_id` can be created against merchant's bank account data. The `recipient_id` can be stored in the DB. 
Alternatively, for the connectors that don't offer this flow, we can offer to store the data in locker.

### Possible Implementation

A recipient creation API call would have to be incorporated in the payment connector create flow, some type changes would be required a as well.

### Have you spent some time checking if this feature request has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/config/config.example.toml b/config/config.example.toml
index 9b855b1e956..5b4c1bbd762 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -698,3 +698,6 @@ public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "p
 
 [user_auth_methods]
 encryption_key = "" # Encryption key used for encrypting data in user_authentication_methods table
+
+[locker_based_open_banking_connectors]
+connector_list = ""
\ No newline at end of file
diff --git a/config/deployments/integration_test.toml b/config/deployments/integration_test.toml
index b8945db5f24..060756cf910 100644
--- a/config/deployments/integration_test.toml
+++ b/config/deployments/integration_test.toml
@@ -347,3 +347,6 @@ keys = "user-agent"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
+
+[locker_based_open_banking_connectors]
+connector_list = ""
diff --git a/config/deployments/production.toml b/config/deployments/production.toml
index acc28a32c5f..ce7e6df9897 100644
--- a/config/deployments/production.toml
+++ b/config/deployments/production.toml
@@ -366,3 +366,6 @@ keys = "user-agent"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
+
+[locker_based_open_banking_connectors]
+connector_list = ""
diff --git a/config/deployments/sandbox.toml b/config/deployments/sandbox.toml
index d776483ebcc..4920ca8589e 100644
--- a/config/deployments/sandbox.toml
+++ b/config/deployments/sandbox.toml
@@ -370,3 +370,6 @@ keys = "user-agent"
 
 [saved_payment_methods]
 sdk_eligible_payment_methods = "card"
+
+[locker_based_open_banking_connectors]
+connector_list = ""
diff --git a/config/development.toml b/config/development.toml
index c35de87e808..2b7cc68e8b5 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -695,3 +695,6 @@ public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "p
 
 [user_auth_methods]
 encryption_key = "A8EF32E029BC3342E54BF2E172A4D7AA43E8EF9D2C3A624A9F04E2EF79DC698F"
+
+[locker_based_open_banking_connectors]
+connector_list = ""
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index c26055f436a..18a3189e4fd 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -351,7 +351,7 @@ sofort = { country = "AT,BE,DE,ES,CH,NL", currency = "CHF,EUR" }
 open_banking_uk = { country = "DE,GB,AT,BE,CY,EE,ES,FI,FR,GR,HR,IE,IT,LT,LU,LV,MT,NL,PT,SI,SK,BG,CZ,DK,HU,NO,PL,RO,SE,AU,BR", currency = "EUR,GBP,DKK,NOK,PLN,SEK,AUD,BRL" }
 
 [pm_filters.razorpay]
-upi_collect = {country = "IN", currency = "INR"}
+upi_collect = { country = "IN", currency = "INR" }
 
 [pm_filters.zen]
 credit = { not_available_flows = { capture_method = "manual" } }
@@ -442,7 +442,7 @@ delay_between_retries_in_milliseconds = 500
 
 [events.kafka]
 brokers = ["localhost:9092"]
-fraud_check_analytics_topic= "hyperswitch-fraud-check-events"
+fraud_check_analytics_topic = "hyperswitch-fraud-check-events"
 intent_analytics_topic = "hyperswitch-payment-intent-events"
 attempt_analytics_topic = "hyperswitch-payment-attempt-events"
 refund_analytics_topic = "hyperswitch-refund-events"
@@ -519,7 +519,7 @@ enabled = false
 global_tenant = { schema = "public", redis_key_prefix = "" }
 
 [multitenancy.tenants]
-public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default"}
+public = { name = "hyperswitch", base_url = "http://localhost:8080", schema = "public", redis_key_prefix = "", clickhouse_database = "default" }
 
 [user_auth_methods]
 encryption_key = "A8EF32E029BC3342E54BF2E172A4D7AA43E8EF9D2C3A624A9F04E2EF79DC698F"
@@ -548,7 +548,10 @@ merchant_name = "HyperSwitch"
 card = "credit,debit"
 
 [payout_method_filters.adyenplatform]
-sepa = { country = "ES,SK,AT,NL,DE,BE,FR,FI,PT,IE,EE,LT,LV,IT,CZ,DE,HU,NO,PL,SE,GB,CH" , currency = "EUR,CZK,DKK,HUF,NOR,PLN,SEK,GBP,CHF" }
+sepa = { country = "ES,SK,AT,NL,DE,BE,FR,FI,PT,IE,EE,LT,LV,IT,CZ,DE,HU,NO,PL,SE,GB,CH", currency = "EUR,CZK,DKK,HUF,NOR,PLN,SEK,GBP,CHF" }
 
 [payout_method_filters.stripe]
 ach = { country = "US", currency = "USD" }
+
+[locker_based_open_banking_connectors]
+connector_list = ""
diff --git a/crates/api_models/src/admin.rs b/crates/api_models/src/admin.rs
index ec9172b7fdf..20129e4b239 100644
--- a/crates/api_models/src/admin.rs
+++ b/crates/api_models/src/admin.rs
@@ -652,6 +652,49 @@ pub struct MerchantConnectorCreate {
 
     #[schema(value_type = Option<ConnectorStatus>, example = "inactive")]
     pub status: Option<api_enums::ConnectorStatus>,
+
+    /// In case the merchant needs to store any additional sensitive data
+    #[schema(value_type = Option<AdditionalMerchantData>)]
+    pub additional_merchant_data: Option<AdditionalMerchantData>,
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum AdditionalMerchantData {
+    OpenBankingRecipientData(MerchantRecipientData),
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum MerchantAccountData {
+    Iban {
+        #[schema(value_type= String)]
+        iban: Secret<String>,
+        name: String,
+        #[schema(value_type= Option<String>)]
+        #[serde(skip_serializing_if = "Option::is_none")]
+        connector_recipient_id: Option<Secret<String>>,
+    },
+    Bacs {
+        #[schema(value_type= String)]
+        account_number: Secret<String>,
+        #[schema(value_type= String)]
+        sort_code: Secret<String>,
+        name: String,
+        #[schema(value_type= Option<String>)]
+        #[serde(skip_serializing_if = "Option::is_none")]
+        connector_recipient_id: Option<Secret<String>>,
+    },
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum MerchantRecipientData {
+    #[schema(value_type= Option<String>)]
+    ConnectorRecipientId(Secret<String>),
+    #[schema(value_type= Option<String>)]
+    WalletId(Secret<String>),
+    AccountData(MerchantAccountData),
 }
 
 // Different patterns of authentication.
@@ -805,6 +848,9 @@ pub struct MerchantConnectorResponse {
 
     #[schema(value_type = ConnectorStatus, example = "inactive")]
     pub status: api_enums::ConnectorStatus,
+
+    #[schema(value_type = Option<AdditionalMerchantData>)]
+    pub additional_merchant_data: Option<AdditionalMerchantData>,
 }
 
 /// Create a new Merchant Connector for the merchant account. The connector could be a payment processor / facilitator / acquirer or specialized services like Fraud / Accounting etc."
diff --git a/crates/common_enums/src/enums.rs b/crates/common_enums/src/enums.rs
index 87d4f5e775a..ca37332cb76 100644
--- a/crates/common_enums/src/enums.rs
+++ b/crates/common_enums/src/enums.rs
@@ -176,6 +176,7 @@ pub enum RoutableConnectors {
     Worldline,
     Worldpay,
     Zen,
+    Plaid,
     Zsl,
 }
 
diff --git a/crates/diesel_models/src/merchant_connector_account.rs b/crates/diesel_models/src/merchant_connector_account.rs
index 680e3dacc85..e01c465d1b4 100644
--- a/crates/diesel_models/src/merchant_connector_account.rs
+++ b/crates/diesel_models/src/merchant_connector_account.rs
@@ -43,6 +43,7 @@ pub struct MerchantConnectorAccount {
     pub applepay_verified_domains: Option<Vec<String>>,
     pub pm_auth_config: Option<serde_json::Value>,
     pub status: storage_enums::ConnectorStatus,
+    pub additional_merchant_data: Option<Encryption>,
     pub connector_wallets_details: Option<Encryption>,
 }
 
@@ -73,6 +74,7 @@ pub struct MerchantConnectorAccountNew {
     pub applepay_verified_domains: Option<Vec<String>>,
     pub pm_auth_config: Option<serde_json::Value>,
     pub status: storage_enums::ConnectorStatus,
+    pub additional_merchant_data: Option<Encryption>,
     pub connector_wallets_details: Option<Encryption>,
 }
 
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 8d51e378fd9..d9f5b299862 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -707,6 +707,7 @@ diesel::table! {
         applepay_verified_domains -> Nullable<Array<Nullable<Text>>>,
         pm_auth_config -> Nullable<Jsonb>,
         status -> ConnectorStatus,
+        additional_merchant_data -> Nullable<Bytea>,
         connector_wallets_details -> Nullable<Bytea>,
     }
 }
diff --git a/crates/kgraph_utils/benches/evaluation.rs b/crates/kgraph_utils/benches/evaluation.rs
index 820e3da42d8..a137a6fd54d 100644
--- a/crates/kgraph_utils/benches/evaluation.rs
+++ b/crates/kgraph_utils/benches/evaluation.rs
@@ -71,6 +71,7 @@ fn build_test_data(
         applepay_verified_domains: None,
         pm_auth_config: None,
         status: api_enums::ConnectorStatus::Inactive,
+        additional_merchant_data: None,
     };
     let config = CountryCurrencyFilter {
         connector_configs: HashMap::new(),
diff --git a/crates/kgraph_utils/src/mca.rs b/crates/kgraph_utils/src/mca.rs
index 74cc096e730..0cbf171495b 100644
--- a/crates/kgraph_utils/src/mca.rs
+++ b/crates/kgraph_utils/src/mca.rs
@@ -759,6 +759,7 @@ mod tests {
             applepay_verified_domains: None,
             pm_auth_config: None,
             status: api_enums::ConnectorStatus::Inactive,
+            additional_merchant_data: None,
         };
 
         let config_map = kgraph_types::CountryCurrencyFilter {
diff --git a/crates/openapi/src/openapi.rs b/crates/openapi/src/openapi.rs
index 82bde980a1f..d2e669175eb 100644
--- a/crates/openapi/src/openapi.rs
+++ b/crates/openapi/src/openapi.rs
@@ -256,6 +256,9 @@ Never share your secret api keys. Keep them guarded and secure.
         api_models::enums::AuthorizationStatus,
         api_models::enums::PaymentMethodStatus,
         api_models::admin::MerchantConnectorCreate,
+        api_models::admin::AdditionalMerchantData,
+        api_models::admin::MerchantRecipientData,
+        api_models::admin::MerchantAccountData,
         api_models::admin::MerchantConnectorUpdate,
         api_models::admin::PrimaryBusinessDetails,
         api_models::admin::FrmConfigs,
diff --git a/crates/pm_auth/src/connector/plaid.rs b/crates/pm_auth/src/connector/plaid.rs
index dfcfbb7eddc..6f91b2cb1a2 100644
--- a/crates/pm_auth/src/connector/plaid.rs
+++ b/crates/pm_auth/src/connector/plaid.rs
@@ -15,7 +15,9 @@ use crate::{
     types::{
         self as auth_types,
         api::{
-            auth_service::{self, BankAccountCredentials, ExchangeToken, LinkToken},
+            auth_service::{
+                self, BankAccountCredentials, ExchangeToken, LinkToken, RecipientCreate,
+            },
             ConnectorCommon, ConnectorCommonExt, ConnectorIntegration,
         },
     },
@@ -89,6 +91,8 @@ impl ConnectorCommon for Plaid {
 }
 
 impl auth_service::AuthService for Plaid {}
+impl auth_service::PaymentInitiationRecipientCreate for Plaid {}
+impl auth_service::PaymentInitiation for Plaid {}
 impl auth_service::AuthServiceLinkToken for Plaid {}
 
 impl ConnectorIntegration<LinkToken, auth_types::LinkTokenRequest, auth_types::LinkTokenResponse>
@@ -338,3 +342,91 @@ impl
         self.build_error_response(res)
     }
 }
+
+impl
+    ConnectorIntegration<
+        RecipientCreate,
+        auth_types::RecipientCreateRequest,
+        auth_types::RecipientCreateResponse,
+    > for Plaid
+{
+    fn get_headers(
+        &self,
+        req: &auth_types::RecipientCreateRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        _req: &auth_types::RecipientCreateRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<String, errors::ConnectorError> {
+        Ok(format!(
+            "{}{}",
+            self.base_url(connectors),
+            "/payment_initiation/recipient/create"
+        ))
+    }
+
+    fn get_request_body(
+        &self,
+        req: &auth_types::RecipientCreateRouterData,
+    ) -> errors::CustomResult<RequestContent, errors::ConnectorError> {
+        let req_obj = plaid::PlaidRecipientCreateRequest::from(req);
+        Ok(RequestContent::Json(Box::new(req_obj)))
+    }
+
+    fn build_request(
+        &self,
+        req: &auth_types::RecipientCreateRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Post)
+                .url(&auth_types::PaymentInitiationRecipientCreateType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(
+                    auth_types::PaymentInitiationRecipientCreateType::get_headers(
+                        self, req, connectors,
+                    )?,
+                )
+                .set_body(
+                    auth_types::PaymentInitiationRecipientCreateType::get_request_body(self, req)?,
+                )
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &auth_types::RecipientCreateRouterData,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::RecipientCreateRouterData, errors::ConnectorError> {
+        let response: plaid::PlaidRecipientCreateResponse = res
+            .response
+            .parse_struct("PlaidRecipientCreateResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        Ok(<auth_types::RecipientCreateRouterData>::from(
+            auth_types::ResponseRouterData {
+                response,
+                data: data.clone(),
+                http_code: res.status_code,
+            },
+        ))
+    }
+    fn get_error_response(
+        &self,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res)
+    }
+}
diff --git a/crates/pm_auth/src/connector/plaid/transformers.rs b/crates/pm_auth/src/connector/plaid/transformers.rs
index 8ce5aca7b83..571f27f5b20 100644
--- a/crates/pm_auth/src/connector/plaid/transformers.rs
+++ b/crates/pm_auth/src/connector/plaid/transformers.rs
@@ -113,6 +113,103 @@ impl TryFrom<&types::ExchangeTokenRouterData> for PlaidExchangeTokenRequest {
     }
 }
 
+#[derive(Debug, Serialize, Eq, PartialEq)]
+pub struct PlaidRecipientCreateRequest {
+    pub name: String,
+    #[serde(flatten)]
+    pub account_data: PlaidRecipientAccountData,
+    pub address: Option<PlaidRecipientCreateAddress>,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidRecipientCreateResponse {
+    pub recipient_id: String,
+}
+
+#[derive(Debug, Serialize, Eq, PartialEq)]
+#[serde(rename_all = "lowercase")]
+pub enum PlaidRecipientAccountData {
+    Iban(Secret<String>),
+    Bacs {
+        sort_code: Secret<String>,
+        account: Secret<String>,
+    },
+}
+
+impl From<&types::RecipientAccountData> for PlaidRecipientAccountData {
+    fn from(item: &types::RecipientAccountData) -> Self {
+        match item {
+            types::RecipientAccountData::Iban(iban) => Self::Iban(iban.clone()),
+            types::RecipientAccountData::Bacs {
+                sort_code,
+                account_number,
+            } => Self::Bacs {
+                sort_code: sort_code.clone(),
+                account: account_number.clone(),
+            },
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Eq, PartialEq)]
+pub struct PlaidRecipientCreateAddress {
+    pub street: String,
+    pub city: String,
+    pub postal_code: String,
+    pub country: String,
+}
+
+impl From<&types::RecipientCreateAddress> for PlaidRecipientCreateAddress {
+    fn from(item: &types::RecipientCreateAddress) -> Self {
+        Self {
+            street: item.street.clone(),
+            city: item.city.clone(),
+            postal_code: item.postal_code.clone(),
+            country: common_enums::CountryAlpha2::to_string(&item.country),
+        }
+    }
+}
+
+impl From<&types::RecipientCreateRouterData> for PlaidRecipientCreateRequest {
+    fn from(item: &types::RecipientCreateRouterData) -> Self {
+        Self {
+            name: item.request.name.clone(),
+            account_data: PlaidRecipientAccountData::from(&item.request.account_data),
+            address: item
+                .request
+                .address
+                .as_ref()
+                .map(PlaidRecipientCreateAddress::from),
+        }
+    }
+}
+
+impl<F, T>
+    From<
+        types::ResponseRouterData<
+            F,
+            PlaidRecipientCreateResponse,
+            T,
+            types::RecipientCreateResponse,
+        >,
+    > for types::PaymentAuthRouterData<F, T, types::RecipientCreateResponse>
+{
+    fn from(
+        item: types::ResponseRouterData<
+            F,
+            PlaidRecipientCreateResponse,
+            T,
+            types::RecipientCreateResponse,
+        >,
+    ) -> Self {
+        Self {
+            response: Ok(types::RecipientCreateResponse {
+                recipient_id: item.response.recipient_id,
+            }),
+            ..item.data
+        }
+    }
+}
 #[derive(Debug, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "snake_case")]
 pub struct PlaidBankAccountCredentialsRequest {
@@ -351,6 +448,7 @@ impl<F, T>
 pub struct PlaidAuthType {
     pub client_id: Secret<String>,
     pub secret: Secret<String>,
+    pub merchant_data: Option<types::MerchantRecipientData>,
 }
 
 impl TryFrom<&types::ConnectorAuthType> for PlaidAuthType {
@@ -360,6 +458,16 @@ impl TryFrom<&types::ConnectorAuthType> for PlaidAuthType {
             types::ConnectorAuthType::BodyKey { client_id, secret } => Ok(Self {
                 client_id: client_id.to_owned(),
                 secret: secret.to_owned(),
+                merchant_data: None,
+            }),
+            types::ConnectorAuthType::OpenBankingAuth {
+                api_key,
+                key1,
+                merchant_data,
+            } => Ok(Self {
+                client_id: api_key.to_owned(),
+                secret: key1.to_owned(),
+                merchant_data: Some(merchant_data.clone()),
             }),
             _ => Err(errors::ConnectorError::FailedToObtainAuthType.into()),
         }
diff --git a/crates/pm_auth/src/types.rs b/crates/pm_auth/src/types.rs
index 2537cdc6a36..7450251f157 100644
--- a/crates/pm_auth/src/types.rs
+++ b/crates/pm_auth/src/types.rs
@@ -2,10 +2,11 @@ pub mod api;
 
 use std::marker::PhantomData;
 
-use api::auth_service::{BankAccountCredentials, ExchangeToken, LinkToken};
-use common_enums::{PaymentMethod, PaymentMethodType};
+use api::auth_service::{BankAccountCredentials, ExchangeToken, LinkToken, RecipientCreate};
+use common_enums::{CountryAlpha2, PaymentMethod, PaymentMethodType};
 use common_utils::{id_type, types};
 use masking::Secret;
+
 #[derive(Debug, Clone)]
 pub struct PaymentAuthRouterData<F, Request, Response> {
     pub flow: PhantomData<F>,
@@ -111,6 +112,38 @@ pub type BankDetailsRouterData = PaymentAuthRouterData<
     BankAccountCredentialsResponse,
 >;
 
+#[derive(Debug, Clone)]
+pub struct RecipientCreateRequest {
+    pub name: String,
+    pub account_data: RecipientAccountData,
+    pub address: Option<RecipientCreateAddress>,
+}
+
+#[derive(Debug, Clone)]
+pub struct RecipientCreateResponse {
+    pub recipient_id: String,
+}
+
+#[derive(Debug, Clone)]
+pub enum RecipientAccountData {
+    Iban(Secret<String>),
+    Bacs {
+        sort_code: Secret<String>,
+        account_number: Secret<String>,
+    },
+}
+
+#[derive(Debug, Clone)]
+pub struct RecipientCreateAddress {
+    pub street: String,
+    pub city: String,
+    pub postal_code: String,
+    pub country: CountryAlpha2,
+}
+
+pub type RecipientCreateRouterData =
+    PaymentAuthRouterData<RecipientCreate, RecipientCreateRequest, RecipientCreateResponse>;
+
 pub type PaymentAuthLinkTokenType =
     dyn api::ConnectorIntegration<LinkToken, LinkTokenRequest, LinkTokenResponse>;
 
@@ -123,6 +156,9 @@ pub type PaymentAuthBankAccountDetailsType = dyn api::ConnectorIntegration<
     BankAccountCredentialsResponse,
 >;
 
+pub type PaymentInitiationRecipientCreateType =
+    dyn api::ConnectorIntegration<RecipientCreate, RecipientCreateRequest, RecipientCreateResponse>;
+
 #[derive(Clone, Debug, strum::EnumString, strum::Display)]
 #[strum(serialize_all = "snake_case")]
 pub enum PaymentMethodAuthConnectors {
@@ -155,12 +191,38 @@ impl ErrorResponse {
     }
 }
 
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+pub enum MerchantAccountData {
+    Iban {
+        iban: Secret<String>,
+        name: String,
+    },
+    Bacs {
+        account_number: Secret<String>,
+        sort_code: Secret<String>,
+        name: String,
+    },
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MerchantRecipientData {
+    ConnectorRecipientId(Secret<String>),
+    WalletId(Secret<String>),
+    AccountData(MerchantAccountData),
+}
+
 #[derive(Default, Debug, Clone, serde::Deserialize)]
 pub enum ConnectorAuthType {
     BodyKey {
         client_id: Secret<String>,
         secret: Secret<String>,
     },
+    OpenBankingAuth {
+        api_key: Secret<String>,
+        key1: Secret<String>,
+        merchant_data: MerchantRecipientData,
+    },
     #[default]
     NoKey,
 }
diff --git a/crates/pm_auth/src/types/api.rs b/crates/pm_auth/src/types/api.rs
index 3684e34ec05..d4edf160985 100644
--- a/crates/pm_auth/src/types/api.rs
+++ b/crates/pm_auth/src/types/api.rs
@@ -10,7 +10,10 @@ use masking::Maskable;
 
 use crate::{
     core::errors::ConnectorError,
-    types::{self as auth_types, api::auth_service::AuthService},
+    types::{
+        self as auth_types,
+        api::auth_service::{AuthService, PaymentInitiation},
+    },
 };
 
 #[async_trait::async_trait]
@@ -125,9 +128,9 @@ where
     }
 }
 
-pub trait AuthServiceConnector: AuthService + Send + Debug {}
+pub trait AuthServiceConnector: AuthService + Send + Debug + PaymentInitiation {}
 
-impl<T: Send + Debug + AuthService> AuthServiceConnector for T {}
+impl<T: Send + Debug + AuthService + PaymentInitiation> AuthServiceConnector for T {}
 
 pub type BoxedPaymentAuthConnector = Box<&'static (dyn AuthServiceConnector + Sync)>;
 
diff --git a/crates/pm_auth/src/types/api/auth_service.rs b/crates/pm_auth/src/types/api/auth_service.rs
index 35d44970d51..498b6bec40e 100644
--- a/crates/pm_auth/src/types/api/auth_service.rs
+++ b/crates/pm_auth/src/types/api/auth_service.rs
@@ -1,6 +1,7 @@
 use crate::types::{
     BankAccountCredentialsRequest, BankAccountCredentialsResponse, ExchangeTokenRequest,
-    ExchangeTokenResponse, LinkTokenRequest, LinkTokenResponse,
+    ExchangeTokenResponse, LinkTokenRequest, LinkTokenResponse, RecipientCreateRequest,
+    RecipientCreateResponse,
 };
 
 pub trait AuthService:
@@ -11,6 +12,8 @@ pub trait AuthService:
 {
 }
 
+pub trait PaymentInitiation: super::ConnectorCommon + PaymentInitiationRecipientCreate {}
+
 #[derive(Debug, Clone)]
 pub struct LinkToken;
 
@@ -38,3 +41,11 @@ pub trait AuthServiceBankAccountCredentials:
 >
 {
 }
+
+#[derive(Debug, Clone)]
+pub struct RecipientCreate;
+
+pub trait PaymentInitiationRecipientCreate:
+    super::ConnectorIntegration<RecipientCreate, RecipientCreateRequest, RecipientCreateResponse>
+{
+}
diff --git a/crates/router/src/configs/secrets_transformers.rs b/crates/router/src/configs/secrets_transformers.rs
index 9f6c797cde8..de53f9c189e 100644
--- a/crates/router/src/configs/secrets_transformers.rs
+++ b/crates/router/src/configs/secrets_transformers.rs
@@ -434,5 +434,6 @@ pub(crate) async fn fetch_raw_secrets(
         multitenancy: conf.multitenancy,
         user_auth_methods,
         decision: conf.decision,
+        locker_based_open_banking_connectors: conf.locker_based_open_banking_connectors,
     }
 }
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index b96784eb714..26685d63583 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -128,6 +128,7 @@ pub struct Settings<S: SecretState> {
     pub saved_payment_methods: EligiblePaymentMethods,
     pub user_auth_methods: SecretStateContainer<UserAuthMethodSettings, S>,
     pub decision: Option<DecisionConfig>,
+    pub locker_based_open_banking_connectors: LockerBasedRecipientConnectorList,
 }
 
 #[derive(Debug, Deserialize, Clone, Default)]
@@ -690,6 +691,13 @@ pub struct ApplePayDecryptConifg {
     pub apple_pay_merchant_cert_key: Secret<String>,
 }
 
+#[derive(Debug, Deserialize, Clone, Default)]
+#[serde(default)]
+pub struct LockerBasedRecipientConnectorList {
+    #[serde(deserialize_with = "deserialize_hashset")]
+    pub connector_list: HashSet<String>,
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 pub struct ConnectorRequestReferenceIdConfig {
     pub merchant_ids_send_payment_id_as_connector_request_id: HashSet<String>,
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index 67b3a05adb7..88f11fdfce8 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -7,7 +7,7 @@ use api_models::{
 use common_utils::{
     date_time,
     ext_traits::{AsyncExt, ConfigExt, Encode, ValueExt},
-    pii,
+    id_type, pii,
 };
 #[cfg(all(feature = "keymanager_create", feature = "olap"))]
 use common_utils::{keymanager, types::keymanager as km_types};
@@ -15,24 +15,25 @@ use diesel_models::configs;
 use error_stack::{report, FutureExt, ResultExt};
 use futures::future::try_join_all;
 use masking::{PeekInterface, Secret};
-use pm_auth::connector::plaid::transformers::PlaidAuthType;
+use pm_auth::{connector::plaid::transformers::PlaidAuthType, types as pm_auth_types};
+use regex::Regex;
 use router_env::metrics::add_attributes;
 use uuid::Uuid;
 
-#[cfg(all(not(feature = "v2"), feature = "olap"))]
-use crate::types::transformers::ForeignFrom;
 use crate::{
     consts,
     core::{
         encryption::transfer_encryption_key,
         errors::{self, RouterResponse, RouterResult, StorageErrorExt},
+        payment_methods::{cards, transformers},
         payments::helpers,
+        pm_auth::helpers::PaymentAuthConnectorDataExt,
         routing::helpers as routing_helpers,
         utils as core_utils,
     },
     db::StorageInterface,
     routes::{metrics, SessionState},
-    services::{self, api as service_api, authentication},
+    services::{self, api as service_api, authentication, pm_auth as payment_initiation_service},
     types::{
         self, api,
         domain::{
@@ -40,11 +41,15 @@ use crate::{
             types::{self as domain_types, AsyncLift},
         },
         storage::{self, enums::MerchantStorageScheme},
-        transformers::ForeignTryFrom,
+        transformers::{ForeignFrom, ForeignTryFrom},
     },
     utils::{self, OptionExt},
 };
 
+const IBAN_MAX_LENGTH: usize = 34;
+const BACS_SORT_CODE_LENGTH: usize = 6;
+const BACS_MAX_ACCOUNT_NUMBER_LENGTH: usize = 8;
+
 #[inline]
 pub fn create_merchant_publishable_key() -> String {
     format!(
@@ -1107,7 +1112,9 @@ pub async fn create_payment_connector(
         api_enums::convert_authentication_connector(req.connector_name.to_string().as_str());
 
     if pm_auth_connector.is_some() {
-        if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth {
+        if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth
+            && req.connector_type != api_enums::ConnectorType::PaymentProcessor
+        {
             return Err(errors::ApiErrorResponse::InvalidRequestData {
                 message: "Invalid connector type given".to_string(),
             }
@@ -1172,6 +1179,30 @@ pub async fn create_payment_connector(
             expected_format: "auth_type and api_key".to_string(),
         })?;
 
+    let merchant_recipient_data = if let Some(data) = &req.additional_merchant_data {
+        Some(
+            process_open_banking_connectors(
+                &state,
+                merchant_id.as_str(),
+                &auth,
+                &req.connector_type,
+                &req.connector_name,
+                types::AdditionalMerchantData::foreign_from(data.clone()),
+            )
+            .await?,
+        )
+    } else {
+        None
+    }
+    .map(|data| {
+        serde_json::to_value(types::AdditionalMerchantData::OpenBankingRecipientData(
+            data,
+        ))
+    })
+    .transpose()
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed to get MerchantRecipientData")?;
+
     validate_auth_and_metadata_type(req.connector_name, &auth, &req.metadata)?;
 
     let frm_configs = get_frm_config_as_secret(req.frm_configs);
@@ -1192,7 +1223,7 @@ pub async fn create_payment_connector(
     let (connector_status, disabled) = validate_status_and_disabled(
         req.status,
         req.disabled,
-        auth,
+        auth.clone(),
         // The validate_status_and_disabled function will use this value only
         // when the status can be active. So we are passing this as fallback.
         api_enums::ConnectorStatus::Active,
@@ -1212,17 +1243,18 @@ pub async fn create_payment_connector(
         }
     }
 
+    let connector_auth = serde_json::to_value(auth)
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed while encoding ConnectorAuthType to serde_json::Value")?;
+    let conn_auth = Secret::new(connector_auth);
+
     let merchant_connector_account = domain::MerchantConnectorAccount {
         merchant_id: merchant_id.to_string(),
         connector_type: req.connector_type,
         connector_name: req.connector_name.to_string(),
         merchant_connector_id: utils::generate_id(consts::ID_LENGTH, "mca"),
         connector_account_details: domain_types::encrypt(
-            req.connector_account_details.ok_or(
-                errors::ApiErrorResponse::MissingRequiredField {
-                    field_name: "connector_account_details",
-                },
-            )?,
+            conn_auth,
             key_store.key.peek(),
         )
         .await
@@ -1256,6 +1288,17 @@ pub async fn create_payment_connector(
         pm_auth_config: req.pm_auth_config.clone(),
         status: connector_status,
         connector_wallets_details: helpers::get_encrypted_apple_pay_connector_wallets_details(&key_store, &req.metadata).await?,
+        additional_merchant_data: if let Some(mcd) =  merchant_recipient_data {
+            Some(domain_types::encrypt(
+                Secret::new(mcd),
+                key_store.key.peek(),
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Unable to encrypt additional_merchant_data")?)
+        } else {
+            None
+        },
     };
 
     let transaction_type = match req.connector_type {
@@ -2577,3 +2620,285 @@ pub fn validate_status_and_disabled(
 
     Ok((connector_status, disabled))
 }
+
+async fn process_open_banking_connectors(
+    state: &SessionState,
+    merchant_id: &str,
+    auth: &types::ConnectorAuthType,
+    connector_type: &api_enums::ConnectorType,
+    connector: &api_enums::Connector,
+    additional_merchant_data: types::AdditionalMerchantData,
+) -> RouterResult<types::MerchantRecipientData> {
+    let new_merchant_data = match additional_merchant_data {
+        types::AdditionalMerchantData::OpenBankingRecipientData(merchant_data) => {
+            if connector_type != &api_enums::ConnectorType::PaymentProcessor {
+                return Err(errors::ApiErrorResponse::InvalidConnectorConfiguration {
+                    config:
+                        "OpenBanking connector for Payment Initiation should be a payment processor"
+                            .to_string(),
+                }
+                .into());
+            }
+            match &merchant_data {
+                types::MerchantRecipientData::AccountData(acc_data) => {
+                    validate_bank_account_data(acc_data)?;
+
+                    let connector_name = api_enums::Connector::to_string(connector);
+
+                    let recipient_creation_not_supported = state
+                        .conf
+                        .locker_based_open_banking_connectors
+                        .connector_list
+                        .contains(connector_name.as_str());
+
+                    let recipient_id = if recipient_creation_not_supported {
+                        locker_recipient_create_call(state, merchant_id, acc_data).await
+                    } else {
+                        connector_recipient_create_call(
+                            state,
+                            merchant_id,
+                            connector_name,
+                            auth,
+                            acc_data,
+                        )
+                        .await
+                    }
+                    .attach_printable("failed to get recipient_id")?;
+
+                    let conn_recipient_id = if recipient_creation_not_supported {
+                        Some(types::RecipientIdType::LockerId(Secret::new(recipient_id)))
+                    } else {
+                        Some(types::RecipientIdType::ConnectorId(Secret::new(
+                            recipient_id,
+                        )))
+                    };
+
+                    let account_data = match &acc_data {
+                        types::MerchantAccountData::Iban { iban, name, .. } => {
+                            types::MerchantAccountData::Iban {
+                                iban: iban.clone(),
+                                name: name.clone(),
+                                connector_recipient_id: conn_recipient_id.clone(),
+                            }
+                        }
+                        types::MerchantAccountData::Bacs {
+                            account_number,
+                            sort_code,
+                            name,
+                            ..
+                        } => types::MerchantAccountData::Bacs {
+                            account_number: account_number.clone(),
+                            sort_code: sort_code.clone(),
+                            name: name.clone(),
+                            connector_recipient_id: conn_recipient_id.clone(),
+                        },
+                    };
+
+                    types::MerchantRecipientData::AccountData(account_data)
+                }
+                _ => merchant_data.clone(),
+            }
+        }
+    };
+
+    Ok(new_merchant_data)
+}
+
+fn validate_bank_account_data(data: &types::MerchantAccountData) -> RouterResult<()> {
+    match data {
+        types::MerchantAccountData::Iban { iban, .. } => {
+            // IBAN check algorithm
+            if iban.peek().len() > IBAN_MAX_LENGTH {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "IBAN length must be up to 34 characters".to_string(),
+                }
+                .into());
+            }
+            let pattern = Regex::new(r"^[A-Z0-9]*$")
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("failed to create regex pattern")?;
+
+            let mut iban = iban.peek().to_string();
+
+            if !pattern.is_match(iban.as_str()) {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "IBAN data must be alphanumeric".to_string(),
+                }
+                .into());
+            }
+
+            // MOD check
+            let first_4 = iban.chars().take(4).collect::<String>();
+            iban.push_str(first_4.as_str());
+            let len = iban.len();
+
+            let rearranged_iban = iban
+                .chars()
+                .rev()
+                .take(len - 4)
+                .collect::<String>()
+                .chars()
+                .rev()
+                .collect::<String>();
+
+            let mut result = String::new();
+
+            rearranged_iban.chars().for_each(|c| {
+                if c.is_ascii_uppercase() {
+                    let digit = (u32::from(c) - u32::from('A')) + 10;
+                    result.push_str(&format!("{:02}", digit));
+                } else {
+                    result.push(c);
+                }
+            });
+
+            let num = result
+                .parse::<u128>()
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("failed to validate IBAN")?;
+
+            if num % 97 != 1 {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Invalid IBAN".to_string(),
+                }
+                .into());
+            }
+
+            Ok(())
+        }
+        types::MerchantAccountData::Bacs {
+            account_number,
+            sort_code,
+            ..
+        } => {
+            if account_number.peek().len() > BACS_MAX_ACCOUNT_NUMBER_LENGTH
+                || sort_code.peek().len() != BACS_SORT_CODE_LENGTH
+            {
+                return Err(errors::ApiErrorResponse::InvalidRequestData {
+                    message: "Invalid BACS numbers".to_string(),
+                }
+                .into());
+            }
+
+            Ok(())
+        }
+    }
+}
+
+async fn connector_recipient_create_call(
+    state: &SessionState,
+    merchant_id: &str,
+    connector_name: String,
+    auth: &types::ConnectorAuthType,
+    data: &types::MerchantAccountData,
+) -> RouterResult<String> {
+    let connector = pm_auth_types::api::PaymentAuthConnectorData::get_connector_by_name(
+        connector_name.as_str(),
+    )?;
+
+    let auth = pm_auth_types::ConnectorAuthType::foreign_try_from(auth.clone())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed while converting ConnectorAuthType")?;
+
+    let connector_integration: pm_auth_types::api::BoxedConnectorIntegration<
+        '_,
+        pm_auth_types::api::auth_service::RecipientCreate,
+        pm_auth_types::RecipientCreateRequest,
+        pm_auth_types::RecipientCreateResponse,
+    > = connector.connector.get_connector_integration();
+
+    let req = match data {
+        types::MerchantAccountData::Iban { iban, name, .. } => {
+            pm_auth_types::RecipientCreateRequest {
+                name: name.clone(),
+                account_data: pm_auth_types::RecipientAccountData::Iban(iban.clone()),
+                address: None,
+            }
+        }
+        types::MerchantAccountData::Bacs {
+            account_number,
+            sort_code,
+            name,
+            ..
+        } => pm_auth_types::RecipientCreateRequest {
+            name: name.clone(),
+            account_data: pm_auth_types::RecipientAccountData::Bacs {
+                sort_code: sort_code.clone(),
+                account_number: account_number.clone(),
+            },
+            address: None,
+        },
+    };
+
+    let router_data = pm_auth_types::RecipientCreateRouterData {
+        flow: std::marker::PhantomData,
+        merchant_id: Some(merchant_id.to_owned()),
+        connector: Some(connector_name),
+        request: req,
+        response: Err(pm_auth_types::ErrorResponse {
+            status_code: http::StatusCode::INTERNAL_SERVER_ERROR.as_u16(),
+            code: consts::NO_ERROR_CODE.to_string(),
+            message: consts::UNSUPPORTED_ERROR_MESSAGE.to_string(),
+            reason: None,
+        }),
+        connector_http_status_code: None,
+        connector_auth_type: auth,
+    };
+
+    let resp = payment_initiation_service::execute_connector_processing_step(
+        state,
+        connector_integration,
+        &router_data,
+        &connector.connector_name,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed while calling recipient create connector api")?;
+
+    let recipient_create_resp =
+        resp.response
+            .map_err(|err| errors::ApiErrorResponse::ExternalConnectorError {
+                code: err.code,
+                message: err.message,
+                connector: connector.connector_name.to_string(),
+                status_code: err.status_code,
+                reason: err.reason,
+            })?;
+
+    let recipient_id = recipient_create_resp.recipient_id;
+
+    Ok(recipient_id)
+}
+
+async fn locker_recipient_create_call(
+    state: &SessionState,
+    merchant_id: &str,
+    data: &types::MerchantAccountData,
+) -> RouterResult<String> {
+    let enc_data = serde_json::to_string(data)
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to convert to MerchantAccountData json to String")?;
+
+    let cust_id = id_type::CustomerId::from(merchant_id.to_string().into())
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to convert to CustomerId")?;
+
+    let payload = transformers::StoreLockerReq::LockerGeneric(transformers::StoreGenericReq {
+        merchant_id,
+        merchant_customer_id: cust_id.clone(),
+        enc_data,
+        ttl: state.conf.locker.ttl_for_storage_in_secs,
+    });
+
+    let store_resp = cards::call_to_locker_hs(
+        state,
+        &payload,
+        &cust_id,
+        api_enums::LockerChoice::HyperswitchCardVault,
+    )
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed to encrypt merchant bank account data")?;
+
+    Ok(store_resp.card_reference)
+}
diff --git a/crates/router/src/core/pm_auth.rs b/crates/router/src/core/pm_auth.rs
index d3393b5b1f7..6eea962b9e5 100644
--- a/crates/router/src/core/pm_auth.rs
+++ b/crates/router/src/core/pm_auth.rs
@@ -209,6 +209,7 @@ impl ForeignTryFrom<&types::ConnectorAuthType> for PlaidAuthType {
                 Ok::<Self, errors::ConnectorError>(Self {
                     client_id: api_key.to_owned(),
                     secret: key1.to_owned(),
+                    merchant_data: None,
                 })
             }
             _ => Err(errors::ConnectorError::FailedToObtainAuthType),
diff --git a/crates/router/src/core/pm_auth/transformers.rs b/crates/router/src/core/pm_auth/transformers.rs
index 8a1369c2e02..d516cab62fe 100644
--- a/crates/router/src/core/pm_auth/transformers.rs
+++ b/crates/router/src/core/pm_auth/transformers.rs
@@ -2,6 +2,36 @@ use pm_auth::types::{self as pm_auth_types};
 
 use crate::{core::errors, types, types::transformers::ForeignTryFrom};
 
+impl From<types::MerchantAccountData> for pm_auth_types::MerchantAccountData {
+    fn from(from: types::MerchantAccountData) -> Self {
+        match from {
+            types::MerchantAccountData::Iban { iban, name, .. } => Self::Iban { iban, name },
+            types::MerchantAccountData::Bacs {
+                account_number,
+                sort_code,
+                name,
+                ..
+            } => Self::Bacs {
+                account_number,
+                sort_code,
+                name,
+            },
+        }
+    }
+}
+
+impl From<types::MerchantRecipientData> for pm_auth_types::MerchantRecipientData {
+    fn from(value: types::MerchantRecipientData) -> Self {
+        match value {
+            types::MerchantRecipientData::ConnectorRecipientId(id) => {
+                Self::ConnectorRecipientId(id)
+            }
+            types::MerchantRecipientData::WalletId(id) => Self::WalletId(id),
+            types::MerchantRecipientData::AccountData(data) => Self::AccountData(data.into()),
+        }
+    }
+}
+
 impl ForeignTryFrom<types::ConnectorAuthType> for pm_auth_types::ConnectorAuthType {
     type Error = errors::ConnectorError;
     fn foreign_try_from(auth_type: types::ConnectorAuthType) -> Result<Self, Self::Error> {
diff --git a/crates/router/src/db/merchant_connector_account.rs b/crates/router/src/db/merchant_connector_account.rs
index a176db26549..567d16ee269 100644
--- a/crates/router/src/db/merchant_connector_account.rs
+++ b/crates/router/src/db/merchant_connector_account.rs
@@ -440,7 +440,7 @@ impl MerchantConnectorAccountInterface for Store {
 
                 #[cfg(feature = "accounts_cache")]
                 // Redact all caches as any of might be used because of backwards compatibility
-                cache::publish_and_redact_multiple(
+                Box::pin(cache::publish_and_redact_multiple(
                     self,
                     [
                         cache::CacheKind::Accounts(
@@ -454,7 +454,7 @@ impl MerchantConnectorAccountInterface for Store {
                         ),
                     ],
                     || update,
-                )
+                ))
                 .await
                 .map_err(|error| {
                     // Returning `DatabaseConnectionError` after logging the actual error because
@@ -784,6 +784,7 @@ impl MerchantConnectorAccountInterface for MockDb {
             pm_auth_config: t.pm_auth_config,
             status: t.status,
             connector_wallets_details: t.connector_wallets_details.map(Encryption::from),
+            additional_merchant_data: t.additional_merchant_data.map(|data| data.into()),
         };
         accounts.push(account.clone());
         account
@@ -991,6 +992,7 @@ mod merchant_connector_account_cache_tests {
                 .await
                 .unwrap(),
             ),
+            additional_merchant_data: None,
         };
 
         db.insert_merchant_connector_account(mca.clone(), &merchant_key)
diff --git a/crates/router/src/types.rs b/crates/router/src/types.rs
index c457ca59df8..cc735612ac3 100644
--- a/crates/router/src/types.rs
+++ b/crates/router/src/types.rs
@@ -12,7 +12,7 @@ pub mod domain;
 #[cfg(feature = "frm")]
 pub mod fraud_check;
 pub mod pm_auth;
-
+use masking::Secret;
 pub mod storage;
 pub mod transformers;
 use std::marker::PhantomData;
@@ -606,6 +606,150 @@ pub struct ResponseRouterData<Flow, R, Request, Response> {
     pub http_code: u16,
 }
 
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+pub enum RecipientIdType {
+    ConnectorId(Secret<String>),
+    LockerId(Secret<String>),
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MerchantAccountData {
+    Iban {
+        iban: Secret<String>,
+        name: String,
+        connector_recipient_id: Option<RecipientIdType>,
+    },
+    Bacs {
+        account_number: Secret<String>,
+        sort_code: Secret<String>,
+        name: String,
+        connector_recipient_id: Option<RecipientIdType>,
+    },
+}
+
+impl ForeignFrom<MerchantAccountData> for api_models::admin::MerchantAccountData {
+    fn foreign_from(from: MerchantAccountData) -> Self {
+        match from {
+            MerchantAccountData::Iban {
+                iban,
+                name,
+                connector_recipient_id,
+            } => Self::Iban {
+                iban,
+                name,
+                connector_recipient_id: match connector_recipient_id {
+                    Some(RecipientIdType::ConnectorId(id)) => Some(id.clone()),
+                    _ => None,
+                },
+            },
+            MerchantAccountData::Bacs {
+                account_number,
+                sort_code,
+                name,
+                connector_recipient_id,
+            } => Self::Bacs {
+                account_number,
+                sort_code,
+                name,
+                connector_recipient_id: match connector_recipient_id {
+                    Some(RecipientIdType::ConnectorId(id)) => Some(id.clone()),
+                    _ => None,
+                },
+            },
+        }
+    }
+}
+
+impl From<api_models::admin::MerchantAccountData> for MerchantAccountData {
+    fn from(from: api_models::admin::MerchantAccountData) -> Self {
+        match from {
+            api_models::admin::MerchantAccountData::Iban {
+                iban,
+                name,
+                connector_recipient_id,
+            } => Self::Iban {
+                iban,
+                name,
+                connector_recipient_id: connector_recipient_id.map(RecipientIdType::ConnectorId),
+            },
+            api_models::admin::MerchantAccountData::Bacs {
+                account_number,
+                sort_code,
+                name,
+                connector_recipient_id,
+            } => Self::Bacs {
+                account_number,
+                sort_code,
+                name,
+                connector_recipient_id: connector_recipient_id.map(RecipientIdType::ConnectorId),
+            },
+        }
+    }
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MerchantRecipientData {
+    ConnectorRecipientId(Secret<String>),
+    WalletId(Secret<String>),
+    AccountData(MerchantAccountData),
+}
+
+impl ForeignFrom<MerchantRecipientData> for api_models::admin::MerchantRecipientData {
+    fn foreign_from(value: MerchantRecipientData) -> Self {
+        match value {
+            MerchantRecipientData::ConnectorRecipientId(id) => Self::ConnectorRecipientId(id),
+            MerchantRecipientData::WalletId(id) => Self::WalletId(id),
+            MerchantRecipientData::AccountData(data) => {
+                Self::AccountData(api_models::admin::MerchantAccountData::foreign_from(data))
+            }
+        }
+    }
+}
+
+impl From<api_models::admin::MerchantRecipientData> for MerchantRecipientData {
+    fn from(value: api_models::admin::MerchantRecipientData) -> Self {
+        match value {
+            api_models::admin::MerchantRecipientData::ConnectorRecipientId(id) => {
+                Self::ConnectorRecipientId(id)
+            }
+            api_models::admin::MerchantRecipientData::WalletId(id) => Self::WalletId(id),
+            api_models::admin::MerchantRecipientData::AccountData(data) => {
+                Self::AccountData(data.into())
+            }
+        }
+    }
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum AdditionalMerchantData {
+    OpenBankingRecipientData(MerchantRecipientData),
+}
+
+impl ForeignFrom<api_models::admin::AdditionalMerchantData> for AdditionalMerchantData {
+    fn foreign_from(value: api_models::admin::AdditionalMerchantData) -> Self {
+        match value {
+            api_models::admin::AdditionalMerchantData::OpenBankingRecipientData(data) => {
+                Self::OpenBankingRecipientData(MerchantRecipientData::from(data))
+            }
+        }
+    }
+}
+
+impl ForeignFrom<AdditionalMerchantData> for api_models::admin::AdditionalMerchantData {
+    fn foreign_from(value: AdditionalMerchantData) -> Self {
+        match value {
+            AdditionalMerchantData::OpenBankingRecipientData(data) => {
+                Self::OpenBankingRecipientData(
+                    api_models::admin::MerchantRecipientData::foreign_from(data),
+                )
+            }
+        }
+    }
+}
+
 impl ForeignFrom<api_models::admin::ConnectorAuthType> for ConnectorAuthType {
     fn foreign_from(value: api_models::admin::ConnectorAuthType) -> Self {
         match value {
diff --git a/crates/router/src/types/domain/merchant_connector_account.rs b/crates/router/src/types/domain/merchant_connector_account.rs
index 6c2f6a06e1e..2cf091eda12 100644
--- a/crates/router/src/types/domain/merchant_connector_account.rs
+++ b/crates/router/src/types/domain/merchant_connector_account.rs
@@ -40,6 +40,7 @@ pub struct MerchantConnectorAccount {
     pub pm_auth_config: Option<serde_json::Value>,
     pub status: enums::ConnectorStatus,
     pub connector_wallets_details: Option<Encryptable<Secret<serde_json::Value>>>,
+    pub additional_merchant_data: Option<Encryptable<Secret<serde_json::Value>>>,
 }
 
 #[derive(Debug)]
@@ -101,6 +102,7 @@ impl behaviour::Conversion for MerchantConnectorAccount {
                 pm_auth_config: self.pm_auth_config,
                 status: self.status,
                 connector_wallets_details: self.connector_wallets_details.map(Encryption::from),
+                additional_merchant_data: self.additional_merchant_data.map(|data| data.into()),
             },
         )
     }
@@ -148,6 +150,17 @@ impl behaviour::Conversion for MerchantConnectorAccount {
                 .change_context(ValidationError::InvalidValue {
                     message: "Failed while decrypting connector wallets details".to_string(),
                 })?,
+            additional_merchant_data: if let Some(data) = other.additional_merchant_data {
+                Some(
+                    Encryptable::decrypt(data, key.peek(), GcmAes256)
+                        .await
+                        .change_context(ValidationError::InvalidValue {
+                            message: "Failed while decrypting additional_merchant_data".to_string(),
+                        })?,
+                )
+            } else {
+                None
+            },
         })
     }
 
@@ -177,6 +190,7 @@ impl behaviour::Conversion for MerchantConnectorAccount {
             pm_auth_config: self.pm_auth_config,
             status: self.status,
             connector_wallets_details: self.connector_wallets_details.map(Encryption::from),
+            additional_merchant_data: self.additional_merchant_data.map(|data| data.into()),
         })
     }
 }
diff --git a/crates/router/src/types/transformers.rs b/crates/router/src/types/transformers.rs
index b06599769f4..d1f01ac8ef7 100644
--- a/crates/router/src/types/transformers.rs
+++ b/crates/router/src/types/transformers.rs
@@ -26,6 +26,7 @@ use crate::{
     },
     services::authentication::get_header_value_by_key,
     types::{
+        self as router_types,
         api::{self as api_types, routing as routing_types},
         storage,
     },
@@ -977,6 +978,19 @@ impl TryFrom<domain::MerchantConnectorAccount> for api_models::admin::MerchantCo
             applepay_verified_domains: item.applepay_verified_domains,
             pm_auth_config: item.pm_auth_config,
             status: item.status,
+            additional_merchant_data: item
+                .additional_merchant_data
+                .map(|data| {
+                    let data = data.into_inner();
+                    serde_json::Value::parse_value::<router_types::AdditionalMerchantData>(
+                        data.expose(),
+                        "AdditionalMerchantData",
+                    )
+                    .attach_printable("Unable to deserialize additional_merchant_data")
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                })
+                .transpose()?
+                .map(api_models::admin::AdditionalMerchantData::foreign_from),
         })
     }
 }
diff --git a/migrations/2024-04-12-100925_mca_additional_merchant_data/down.sql b/migrations/2024-04-12-100925_mca_additional_merchant_data/down.sql
new file mode 100644
index 00000000000..7614b52872d
--- /dev/null
+++ b/migrations/2024-04-12-100925_mca_additional_merchant_data/down.sql
@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE merchant_connector_account DROP COLUMN IF EXISTS additional_merchant_data;
\ No newline at end of file
diff --git a/migrations/2024-04-12-100925_mca_additional_merchant_data/up.sql b/migrations/2024-04-12-100925_mca_additional_merchant_data/up.sql
new file mode 100644
index 00000000000..4b48796785e
--- /dev/null
+++ b/migrations/2024-04-12-100925_mca_additional_merchant_data/up.sql
@@ -0,0 +1,2 @@
+-- Your SQL goes here
+ALTER TABLE merchant_connector_account ADD COLUMN IF NOT EXISTS additional_merchant_data BYTEA DEFAULT NULL;
\ No newline at end of file

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
- Added connector call in payment connector create to facilitate merchant recipient creation at connector end.
- Added alternative locker call for creation of recipient at Hyperswitch's end in case the connector does not support it.
- Added validation for merchant bank account data


### Additional Changes

- [x] This PR modifies the API contract
- [x] This PR modifies the database schema
- [x] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## 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).
-->


## 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)?
-->
1. Create an MCA account with `connector_type` as `payment_processor` and an open banking connector
```
curl --location --request POST 'http://localhost:8080/account/merchant_1720607250/connectors' \
--header 'api-key: test_admin' \
--header 'Content-Type: application/json' \
--data-raw '{
    "connector_type": "payment_processor",
    "connector_name": "plaid",
    "connector_label": "plaid3",
    "connector_account_details": {
        "auth_type": "BodyKey",
        "api_key": "Some_key",
        "key1": "Some_key"
    },
    "test_mode": false,
    "disabled": false,
    "payment_methods_enabled": [
        {
            "payment_method": "bank_redirect",
            "payment_method_types": [
                {
                    "payment_method_type": "eps",
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    }
                },
                {
                    "payment_method_type": "giropay",
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    }
                },
                {
                    "payment_method_type": "ideal",
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    }
                }
            ]
        },
        {
            "payment_method": "bank_debit",
            "payment_method_types": [
                {
                    "payment_method_type": "ach",
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "minimum_amount": 0,
                    "maximum_amount": 10000
                }
            ]
        },
        {
            "payment_method": "wallet",
            "payment_method_types": [
                {
                    "payment_method_type": "apple_pay",
                    "payment_experience": "invoke_sdk_client",
                    "card_networks": null,
                    "accepted_countries": null,
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "USD",
                            "GBP",
                            "CAD",
                            "AUD",
                            "EUR"
                        ]
                    }
                }
            ]
        }
    ],
    "additional_merchant_data": {
         "open_banking_recipient_data": {
            "account_data": {
                "bacs": {
                    "sort_code": "200000",
                    "account_number": "55779911",
                    "name": "merchant_name"
                }
            }
        }
    },
    "metadata": {
        "city": "NY",
        "unit": "246"
    }
}'
```
Successful Response should look like this - 
```
{
    "connector_type": "payment_processor",
    "connector_name": "plaid",
    "connector_label": "plaid3",
    "merchant_connector_id": "mca_c0iTEayEJMViYwEhTCOG",
    "profile_id": "pro_UgP3FfqWGhfPLuUqYtJF",
    "connector_account_details": {
        "auth_type": "BodyKey",
      "api_key": "Some_key",
        "key1": "Some_key"
    },
    "payment_methods_enabled": [
        {
            "payment_method": "bank_redirect",
            "payment_method_types": [
                {
                    "payment_method_type": "eps",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "giropay",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "ideal",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        },
        {
            "payment_method": "bank_debit",
            "payment_method_types": [
                {
                    "payment_method_type": "ach",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": 0,
                    "maximum_amount": 10000,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        },
        {
            "payment_method": "wallet",
            "payment_method_types": [
                {
                    "payment_method_type": "apple_pay",
                    "payment_experience": "invoke_sdk_client",
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "USD",
                            "GBP",
                            "CAD",
                            "AUD",
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        }
    ],
    "connector_webhook_details": null,
    "metadata": {
        "city": "NY",
        "unit": "246"
    },
    "test_mode": false,
    "disabled": false,
    "frm_configs": null,
    "business_country": null,
    "business_label": null,
    "business_sub_label": null,
    "applepay_verified_domains": null,
    "pm_auth_config": null,
    "status": "active",
    "additional_merchant_data": {
        "open_banking_recipient_data": {
            "account_data": {
                "bacs": {
                    "account_number": "55779911",
                    "sort_code": "200000",
                    "name": "merchant_name",
                    "connector_recipient_id": "recipient-id-sandbox-45beb880-9e3d-4ae3-927b-052f5610fd74"
                }
            }
        }
    }
}
```
Response when the connector does not support recipient creation (locker call to be made in this)
```
{
    "connector_type": "payment_processor",
    "connector_name": "plaid",
    "connector_label": "plaid3",
    "merchant_connector_id": "mca_6bIUIrWNSbqssRO89Zxx",
    "profile_id": "pro_UgP3FfqWGhfPLuUqYtJF",
    "connector_account_details": {
        "auth_type": "BodyKey",
       "api_key": "Some_key",
        "key1": "Some_key"
    },
    "payment_methods_enabled": [
        {
            "payment_method": "bank_redirect",
            "payment_method_types": [
                {
                    "payment_method_type": "eps",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "giropay",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "ideal",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": null,
                    "maximum_amount": null,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        },
        {
            "payment_method": "bank_debit",
            "payment_method_types": [
                {
                    "payment_method_type": "ach",
                    "payment_experience": null,
                    "card_networks": null,
                    "accepted_currencies": null,
                    "accepted_countries": null,
                    "minimum_amount": 0,
                    "maximum_amount": 10000,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        },
        {
            "payment_method": "wallet",
            "payment_method_types": [
                {
                    "payment_method_type": "apple_pay",
                    "payment_experience": "invoke_sdk_client",
                    "card_networks": null,
                    "accepted_currencies": {
                        "type": "enable_only",
                        "list": [
                            "USD",
                            "GBP",
                            "CAD",
                            "AUD",
                            "EUR"
                        ]
                    },
                    "accepted_countries": null,
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        }
    ],
    "connector_webhook_details": null,
    "metadata": {
        "city": "NY",
        "unit": "246"
    },
    "test_mode": false,
    "disabled": false,
    "frm_configs": null,
    "business_country": null,
    "business_label": null,
    "business_sub_label": null,
    "applepay_verified_domains": null,
    "pm_auth_config": null,
    "status": "active",
    "additional_merchant_data": {
        "open_banking_recipient_data": {
            "account_data": {
                "bacs": {
                    "account_number": "55779911",
                    "sort_code": "200000",
                    "name": "merchant_name"
                }
            }
        }
    }
}
```


## 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`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
