Bug: Update Schema and domain models for Payment Methods v2



### Feature Description

Have to change the schema for corresponding DB changes in `payment_methods` table for v2

### Possible Implementation
Will need to irun separate migrations for v2
Based on need, can implement domain models for Payment Methods as well
diff --git a/crates/diesel_models/Cargo.toml b/crates/diesel_models/Cargo.toml
index 45582cc0ad9..6b288899778 100644
--- a/crates/diesel_models/Cargo.toml
+++ b/crates/diesel_models/Cargo.toml
@@ -14,6 +14,7 @@ v1 = []
 v2 = []
 customer_v2 = []
 payment_v2 = []
+payment_methods_v2 = []
 
 [dependencies]
 async-bb8-diesel = { git = "https://github.com/jarnura/async-bb8-diesel", rev = "53b4ab901aab7635c8215fd1c2d542c8db443094" }
diff --git a/crates/diesel_models/src/kv.rs b/crates/diesel_models/src/kv.rs
index 3361b82f89e..c8940c4c6b2 100644
--- a/crates/diesel_models/src/kv.rs
+++ b/crates/diesel_models/src/kv.rs
@@ -124,11 +124,19 @@ impl DBOperation {
                 Updateable::PayoutAttemptUpdate(a) => DBResult::PayoutAttempt(Box::new(
                     a.orig.update_with_attempt_id(conn, a.update_data).await?,
                 )),
+                #[cfg(all(
+                    any(feature = "v1", feature = "v2"),
+                    not(feature = "payment_methods_v2")
+                ))]
                 Updateable::PaymentMethodUpdate(v) => DBResult::PaymentMethod(Box::new(
                     v.orig
                         .update_with_payment_method_id(conn, v.update_data)
                         .await?,
                 )),
+                #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+                Updateable::PaymentMethodUpdate(v) => DBResult::PaymentMethod(Box::new(
+                    v.orig.update_with_id(conn, v.update_data).await?,
+                )),
                 Updateable::MandateUpdate(m) => DBResult::Mandate(Box::new(
                     Mandate::update_by_merchant_id_mandate_id(
                         conn,
diff --git a/crates/diesel_models/src/payment_method.rs b/crates/diesel_models/src/payment_method.rs
index b54824ad935..c56ab8a0b2b 100644
--- a/crates/diesel_models/src/payment_method.rs
+++ b/crates/diesel_models/src/payment_method.rs
@@ -1,12 +1,26 @@
 use common_enums::MerchantStorageScheme;
 use common_utils::{encryption::Encryption, pii};
 use diesel::{AsChangeset, Identifiable, Insertable, Queryable, Selectable};
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 use masking::Secret;
 use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 use crate::{enums as storage_enums, schema::payment_methods};
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+use crate::{enums as storage_enums, schema_v2::payment_methods};
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 #[derive(
     Clone, Debug, Eq, PartialEq, Identifiable, Queryable, Selectable, Serialize, Deserialize,
 )]
@@ -44,8 +58,56 @@ pub struct PaymentMethod {
     pub client_secret: Option<String>,
     pub payment_method_billing_address: Option<Encryption>,
     pub updated_by: Option<String>,
+    pub version: common_enums::ApiVersion,
 }
 
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+#[derive(
+    Clone, Debug, Eq, PartialEq, Identifiable, Queryable, Selectable, Serialize, Deserialize,
+)]
+#[diesel(table_name = payment_methods, primary_key(id), check_for_backend(diesel::pg::Pg))]
+pub struct PaymentMethod {
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified: PrimitiveDateTime,
+    pub payment_method: Option<storage_enums::PaymentMethod>,
+    pub payment_method_type: Option<storage_enums::PaymentMethodType>,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub payment_method_data: Option<Encryption>,
+    pub locker_id: Option<String>,
+    pub last_used_at: PrimitiveDateTime,
+    pub connector_mandate_details: Option<pii::SecretSerdeValue>,
+    pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub status: storage_enums::PaymentMethodStatus,
+    pub network_transaction_id: Option<String>,
+    pub client_secret: Option<String>,
+    pub payment_method_billing_address: Option<Encryption>,
+    pub updated_by: Option<String>,
+    pub locker_fingerprint_id: Option<String>,
+    pub id: String,
+    pub version: common_enums::ApiVersion,
+}
+
+impl PaymentMethod {
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
+    pub fn get_id(&self) -> &String {
+        &self.payment_method_id
+    }
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    pub fn get_id(&self) -> &String {
+        &self.id
+    }
+}
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 #[derive(
     Clone, Debug, Eq, PartialEq, Insertable, router_derive::DebugAsDisplay, Serialize, Deserialize,
 )]
@@ -81,12 +143,54 @@ pub struct PaymentMethodNew {
     pub client_secret: Option<String>,
     pub payment_method_billing_address: Option<Encryption>,
     pub updated_by: Option<String>,
+    pub version: common_enums::ApiVersion,
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+#[derive(
+    Clone, Debug, Eq, PartialEq, Insertable, router_derive::DebugAsDisplay, Serialize, Deserialize,
+)]
+#[diesel(table_name = payment_methods)]
+pub struct PaymentMethodNew {
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub payment_method: Option<storage_enums::PaymentMethod>,
+    pub payment_method_type: Option<storage_enums::PaymentMethodType>,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified: PrimitiveDateTime,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub payment_method_data: Option<Encryption>,
+    pub locker_id: Option<String>,
+    pub last_used_at: PrimitiveDateTime,
+    pub connector_mandate_details: Option<pii::SecretSerdeValue>,
+    pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub status: storage_enums::PaymentMethodStatus,
+    pub network_transaction_id: Option<String>,
+    pub client_secret: Option<String>,
+    pub payment_method_billing_address: Option<Encryption>,
+    pub updated_by: Option<String>,
+    pub locker_fingerprint_id: Option<String>,
+    pub id: String,
+    pub version: common_enums::ApiVersion,
 }
 
 impl PaymentMethodNew {
     pub fn update_storage_scheme(&mut self, storage_scheme: MerchantStorageScheme) {
         self.updated_by = Some(storage_scheme.to_string());
     }
+
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
+    pub fn get_id(&self) -> &String {
+        &self.payment_method_id
+    }
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    pub fn get_id(&self) -> &String {
+        &self.id
+    }
 }
 
 #[derive(Debug, Eq, PartialEq, Deserialize, Serialize)]
@@ -95,6 +199,10 @@ pub struct TokenizeCoreWorkflow {
     pub pm: storage_enums::PaymentMethod,
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 #[derive(Debug, Serialize, Deserialize)]
 pub enum PaymentMethodUpdate {
     MetadataUpdateAndLastUsed {
@@ -131,6 +239,42 @@ pub enum PaymentMethodUpdate {
     },
 }
 
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+#[derive(Debug, Serialize, Deserialize)]
+pub enum PaymentMethodUpdate {
+    MetadataUpdateAndLastUsed {
+        metadata: Option<pii::SecretSerdeValue>,
+        last_used_at: PrimitiveDateTime,
+    },
+    UpdatePaymentMethodDataAndLastUsed {
+        payment_method_data: Option<Encryption>,
+        last_used_at: PrimitiveDateTime,
+    },
+    PaymentMethodDataUpdate {
+        payment_method_data: Option<Encryption>,
+    },
+    LastUsedUpdate {
+        last_used_at: PrimitiveDateTime,
+    },
+    NetworkTransactionIdAndStatusUpdate {
+        network_transaction_id: Option<String>,
+        status: Option<storage_enums::PaymentMethodStatus>,
+    },
+    StatusUpdate {
+        status: Option<storage_enums::PaymentMethodStatus>,
+    },
+    AdditionalDataUpdate {
+        payment_method_data: Option<Encryption>,
+        status: Option<storage_enums::PaymentMethodStatus>,
+        locker_id: Option<String>,
+        payment_method: Option<storage_enums::PaymentMethod>,
+        payment_method_type: Option<storage_enums::PaymentMethodType>,
+    },
+    ConnectorMandateDetailsUpdate {
+        connector_mandate_details: Option<pii::SecretSerdeValue>,
+    },
+}
+
 impl PaymentMethodUpdate {
     pub fn convert_to_payment_method_update(
         self,
@@ -142,9 +286,64 @@ impl PaymentMethodUpdate {
     }
 }
 
-#[derive(
-    Clone, Debug, Default, AsChangeset, router_derive::DebugAsDisplay, Serialize, Deserialize,
-)]
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+#[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay, Serialize, Deserialize)]
+#[diesel(table_name = payment_methods)]
+pub struct PaymentMethodUpdateInternal {
+    metadata: Option<pii::SecretSerdeValue>,
+    payment_method_data: Option<Encryption>,
+    last_used_at: Option<PrimitiveDateTime>,
+    network_transaction_id: Option<String>,
+    status: Option<storage_enums::PaymentMethodStatus>,
+    locker_id: Option<String>,
+    payment_method: Option<storage_enums::PaymentMethod>,
+    connector_mandate_details: Option<pii::SecretSerdeValue>,
+    updated_by: Option<String>,
+    payment_method_type: Option<storage_enums::PaymentMethodType>,
+    last_modified: PrimitiveDateTime,
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+impl PaymentMethodUpdateInternal {
+    pub fn create_payment_method(self, source: PaymentMethod) -> PaymentMethod {
+        let metadata = self.metadata;
+
+        PaymentMethod { metadata, ..source }
+    }
+
+    pub fn apply_changeset(self, source: PaymentMethod) -> PaymentMethod {
+        let Self {
+            metadata,
+            payment_method_data,
+            last_used_at,
+            network_transaction_id,
+            status,
+            connector_mandate_details,
+            updated_by,
+            ..
+        } = self;
+
+        PaymentMethod {
+            metadata: metadata.map_or(source.metadata, Some),
+            payment_method_data: payment_method_data.map_or(source.payment_method_data, Some),
+            last_used_at: last_used_at.unwrap_or(source.last_used_at),
+            network_transaction_id: network_transaction_id
+                .map_or(source.network_transaction_id, Some),
+            status: status.unwrap_or(source.status),
+            connector_mandate_details: connector_mandate_details
+                .map_or(source.connector_mandate_details, Some),
+            updated_by: updated_by.map_or(source.updated_by, Some),
+            last_modified: common_utils::date_time::now(),
+            ..source
+        }
+    }
+}
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
+#[derive(Clone, Debug, AsChangeset, router_derive::DebugAsDisplay, Serialize, Deserialize)]
 #[diesel(table_name = payment_methods)]
 pub struct PaymentMethodUpdateInternal {
     metadata: Option<serde_json::Value>,
@@ -158,8 +357,13 @@ pub struct PaymentMethodUpdateInternal {
     updated_by: Option<String>,
     payment_method_type: Option<storage_enums::PaymentMethodType>,
     payment_method_issuer: Option<String>,
+    last_modified: PrimitiveDateTime,
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 impl PaymentMethodUpdateInternal {
     pub fn create_payment_method(self, source: PaymentMethod) -> PaymentMethod {
         let metadata = self.metadata.map(Secret::new);
@@ -189,11 +393,16 @@ impl PaymentMethodUpdateInternal {
             connector_mandate_details: connector_mandate_details
                 .map_or(source.connector_mandate_details, Some),
             updated_by: updated_by.map_or(source.updated_by, Some),
+            last_modified: common_utils::date_time::now(),
             ..source
         }
     }
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
     fn from(payment_method_update: PaymentMethodUpdate) -> Self {
         match payment_method_update {
@@ -212,6 +421,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
             },
             PaymentMethodUpdate::PaymentMethodDataUpdate {
                 payment_method_data,
@@ -227,6 +437,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
             },
             PaymentMethodUpdate::LastUsedUpdate { last_used_at } => Self {
                 metadata: None,
@@ -240,6 +451,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
             },
             PaymentMethodUpdate::UpdatePaymentMethodDataAndLastUsed {
                 payment_method_data,
@@ -256,6 +468,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
             },
             PaymentMethodUpdate::NetworkTransactionIdAndStatusUpdate {
                 network_transaction_id,
@@ -272,6 +485,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
             },
             PaymentMethodUpdate::StatusUpdate { status } => Self {
                 metadata: None,
@@ -285,6 +499,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
             },
             PaymentMethodUpdate::AdditionalDataUpdate {
                 payment_method_data,
@@ -305,6 +520,7 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer,
                 payment_method_type,
+                last_modified: common_utils::date_time::now(),
             },
             PaymentMethodUpdate::ConnectorMandateDetailsUpdate {
                 connector_mandate_details,
@@ -320,11 +536,147 @@ impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
                 updated_by: None,
                 payment_method_issuer: None,
                 payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
             },
         }
     }
 }
 
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+impl From<PaymentMethodUpdate> for PaymentMethodUpdateInternal {
+    fn from(payment_method_update: PaymentMethodUpdate) -> Self {
+        match payment_method_update {
+            PaymentMethodUpdate::MetadataUpdateAndLastUsed {
+                metadata,
+                last_used_at,
+            } => Self {
+                metadata,
+                payment_method_data: None,
+                last_used_at: Some(last_used_at),
+                network_transaction_id: None,
+                status: None,
+                locker_id: None,
+                payment_method: None,
+                connector_mandate_details: None,
+                updated_by: None,
+                payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
+            },
+            PaymentMethodUpdate::PaymentMethodDataUpdate {
+                payment_method_data,
+            } => Self {
+                metadata: None,
+                payment_method_data,
+                last_used_at: None,
+                network_transaction_id: None,
+                status: None,
+                locker_id: None,
+                payment_method: None,
+                connector_mandate_details: None,
+                updated_by: None,
+                payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
+            },
+            PaymentMethodUpdate::LastUsedUpdate { last_used_at } => Self {
+                metadata: None,
+                payment_method_data: None,
+                last_used_at: Some(last_used_at),
+                network_transaction_id: None,
+                status: None,
+                locker_id: None,
+                payment_method: None,
+                connector_mandate_details: None,
+                updated_by: None,
+                payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
+            },
+            PaymentMethodUpdate::UpdatePaymentMethodDataAndLastUsed {
+                payment_method_data,
+                last_used_at,
+            } => Self {
+                metadata: None,
+                payment_method_data,
+                last_used_at: Some(last_used_at),
+                network_transaction_id: None,
+                status: None,
+                locker_id: None,
+                payment_method: None,
+                connector_mandate_details: None,
+                updated_by: None,
+                payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
+            },
+            PaymentMethodUpdate::NetworkTransactionIdAndStatusUpdate {
+                network_transaction_id,
+                status,
+            } => Self {
+                metadata: None,
+                payment_method_data: None,
+                last_used_at: None,
+                network_transaction_id,
+                status,
+                locker_id: None,
+                payment_method: None,
+                connector_mandate_details: None,
+                updated_by: None,
+                payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
+            },
+            PaymentMethodUpdate::StatusUpdate { status } => Self {
+                metadata: None,
+                payment_method_data: None,
+                last_used_at: None,
+                network_transaction_id: None,
+                status,
+                locker_id: None,
+                payment_method: None,
+                connector_mandate_details: None,
+                updated_by: None,
+                payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
+            },
+            PaymentMethodUpdate::AdditionalDataUpdate {
+                payment_method_data,
+                status,
+                locker_id,
+                payment_method,
+                payment_method_type,
+            } => Self {
+                metadata: None,
+                payment_method_data,
+                last_used_at: None,
+                network_transaction_id: None,
+                status,
+                locker_id,
+                payment_method,
+                connector_mandate_details: None,
+                updated_by: None,
+                payment_method_type,
+                last_modified: common_utils::date_time::now(),
+            },
+            PaymentMethodUpdate::ConnectorMandateDetailsUpdate {
+                connector_mandate_details,
+            } => Self {
+                metadata: None,
+                payment_method_data: None,
+                last_used_at: None,
+                status: None,
+                locker_id: None,
+                payment_method: None,
+                connector_mandate_details,
+                network_transaction_id: None,
+                updated_by: None,
+                payment_method_type: None,
+                last_modified: common_utils::date_time::now(),
+            },
+        }
+    }
+}
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 impl From<&PaymentMethodNew> for PaymentMethod {
     fn from(payment_method_new: &PaymentMethodNew) -> Self {
         Self {
@@ -360,6 +712,37 @@ impl From<&PaymentMethodNew> for PaymentMethod {
             payment_method_billing_address: payment_method_new
                 .payment_method_billing_address
                 .clone(),
+            version: payment_method_new.version,
+        }
+    }
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+impl From<&PaymentMethodNew> for PaymentMethod {
+    fn from(payment_method_new: &PaymentMethodNew) -> Self {
+        Self {
+            customer_id: payment_method_new.customer_id.clone(),
+            merchant_id: payment_method_new.merchant_id.clone(),
+            locker_id: payment_method_new.locker_id.clone(),
+            created_at: payment_method_new.created_at,
+            last_modified: payment_method_new.last_modified,
+            payment_method: payment_method_new.payment_method,
+            payment_method_type: payment_method_new.payment_method_type,
+            metadata: payment_method_new.metadata.clone(),
+            payment_method_data: payment_method_new.payment_method_data.clone(),
+            last_used_at: payment_method_new.last_used_at,
+            connector_mandate_details: payment_method_new.connector_mandate_details.clone(),
+            customer_acceptance: payment_method_new.customer_acceptance.clone(),
+            status: payment_method_new.status,
+            network_transaction_id: payment_method_new.network_transaction_id.clone(),
+            client_secret: payment_method_new.client_secret.clone(),
+            updated_by: payment_method_new.updated_by.clone(),
+            payment_method_billing_address: payment_method_new
+                .payment_method_billing_address
+                .clone(),
+            id: payment_method_new.id.clone(),
+            locker_fingerprint_id: payment_method_new.locker_fingerprint_id.clone(),
+            version: payment_method_new.version,
         }
     }
 }
diff --git a/crates/diesel_models/src/query/payment_method.rs b/crates/diesel_models/src/query/payment_method.rs
index 90bf455294d..1684356480e 100644
--- a/crates/diesel_models/src/query/payment_method.rs
+++ b/crates/diesel_models/src/query/payment_method.rs
@@ -1,15 +1,36 @@
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 use async_bb8_diesel::AsyncRunQueryDsl;
-use diesel::{
-    associations::HasTable, debug_query, pg::Pg, BoolExpressionMethods, ExpressionMethods,
-    QueryDsl, Table,
-};
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
+use diesel::Table;
+use diesel::{associations::HasTable, BoolExpressionMethods, ExpressionMethods};
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
+use diesel::{debug_query, pg::Pg, QueryDsl};
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 use error_stack::ResultExt;
 
 use super::generics;
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
+use crate::schema::payment_methods::dsl;
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+use crate::schema_v2::payment_methods::dsl::{self, id as pm_id};
 use crate::{
     enums as storage_enums, errors,
     payment_method::{self, PaymentMethod, PaymentMethodNew},
-    schema::payment_methods::dsl,
     PgPooledConn, StorageResult,
 };
 
@@ -19,6 +40,10 @@ impl PaymentMethodNew {
     }
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 impl PaymentMethod {
     pub async fn delete_by_payment_method_id(
         conn: &PgPooledConn,
@@ -173,3 +198,63 @@ impl PaymentMethod {
         }
     }
 }
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+impl PaymentMethod {
+    pub async fn find_by_id(conn: &PgPooledConn, id: &str) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(conn, pm_id.eq(id.to_owned()))
+            .await
+    }
+
+    pub async fn find_by_customer_id_merchant_id_status(
+        conn: &PgPooledConn,
+        customer_id: &common_utils::id_type::CustomerId,
+        merchant_id: &common_utils::id_type::MerchantId,
+        status: storage_enums::PaymentMethodStatus,
+        limit: Option<i64>,
+    ) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<<Self as HasTable>::Table, _, _, _>(
+            conn,
+            dsl::customer_id
+                .eq(customer_id.to_owned())
+                .and(dsl::merchant_id.eq(merchant_id.to_owned()))
+                .and(dsl::status.eq(status)),
+            limit,
+            None,
+            Some(dsl::last_used_at.desc()),
+        )
+        .await
+    }
+
+    pub async fn update_with_id(
+        self,
+        conn: &PgPooledConn,
+        payment_method: payment_method::PaymentMethodUpdateInternal,
+    ) -> StorageResult<Self> {
+        match generics::generic_update_with_unique_predicate_get_result::<
+            <Self as HasTable>::Table,
+            _,
+            _,
+            _,
+        >(conn, pm_id.eq(self.id.to_owned()), payment_method)
+        .await
+        {
+            Err(error) => match error.current_context() {
+                errors::DatabaseError::NoFieldsToUpdate => Ok(self),
+                _ => Err(error),
+            },
+            result => result,
+        }
+    }
+
+    pub async fn find_by_fingerprint_id(
+        conn: &PgPooledConn,
+        fingerprint_id: &str,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::locker_fingerprint_id.eq(fingerprint_id.to_owned()),
+        )
+        .await
+    }
+}
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index e0f2b9a0bd2..7927dfee106 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -1002,6 +1002,7 @@ diesel::table! {
         payment_method_billing_address -> Nullable<Bytea>,
         #[max_length = 64]
         updated_by -> Nullable<Varchar>,
+        version -> ApiVersion,
     }
 }
 
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index 410c541441b..41bf563a3a3 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -933,39 +933,16 @@ diesel::table! {
     use diesel::sql_types::*;
     use crate::enums::diesel_exports::*;
 
-    payment_methods (payment_method_id) {
-        id -> Int4,
+    payment_methods (id) {
         #[max_length = 64]
         customer_id -> Varchar,
         #[max_length = 64]
         merchant_id -> Varchar,
-        #[max_length = 64]
-        payment_method_id -> Varchar,
-        accepted_currency -> Nullable<Array<Nullable<Currency>>>,
-        #[max_length = 32]
-        scheme -> Nullable<Varchar>,
-        #[max_length = 128]
-        token -> Nullable<Varchar>,
-        #[max_length = 255]
-        cardholder_name -> Nullable<Varchar>,
-        #[max_length = 64]
-        issuer_name -> Nullable<Varchar>,
-        #[max_length = 64]
-        issuer_country -> Nullable<Varchar>,
-        payer_country -> Nullable<Array<Nullable<Text>>>,
-        is_stored -> Nullable<Bool>,
-        #[max_length = 32]
-        swift_code -> Nullable<Varchar>,
-        #[max_length = 128]
-        direct_debit_token -> Nullable<Varchar>,
         created_at -> Timestamp,
         last_modified -> Timestamp,
         payment_method -> Nullable<Varchar>,
         #[max_length = 64]
         payment_method_type -> Nullable<Varchar>,
-        #[max_length = 128]
-        payment_method_issuer -> Nullable<Varchar>,
-        payment_method_issuer_code -> Nullable<PaymentMethodIssuerCode>,
         metadata -> Nullable<Json>,
         payment_method_data -> Nullable<Bytea>,
         #[max_length = 64]
@@ -982,6 +959,11 @@ diesel::table! {
         payment_method_billing_address -> Nullable<Bytea>,
         #[max_length = 64]
         updated_by -> Nullable<Varchar>,
+        #[max_length = 64]
+        locker_fingerprint_id -> Nullable<Varchar>,
+        #[max_length = 64]
+        id -> Varchar,
+        version -> ApiVersion,
     }
 }
 
diff --git a/crates/hyperswitch_domain_models/Cargo.toml b/crates/hyperswitch_domain_models/Cargo.toml
index f84d065fd9d..e2ae011c828 100644
--- a/crates/hyperswitch_domain_models/Cargo.toml
+++ b/crates/hyperswitch_domain_models/Cargo.toml
@@ -17,6 +17,7 @@ v2 = ["api_models/v2", "diesel_models/v2"]
 v1 = ["api_models/v1", "diesel_models/v1"]
 customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2"]
 payment_v2 = ["api_models/payment_v2", "diesel_models/payment_v2"]
+payment_methods_v2 = ["api_models/payment_methods_v2", "diesel_models/payment_methods_v2"]
 
 [dependencies]
 # First party deps
diff --git a/crates/hyperswitch_domain_models/src/lib.rs b/crates/hyperswitch_domain_models/src/lib.rs
index 85f35116910..54920dee6ba 100644
--- a/crates/hyperswitch_domain_models/src/lib.rs
+++ b/crates/hyperswitch_domain_models/src/lib.rs
@@ -10,6 +10,7 @@ pub mod merchant_connector_account;
 pub mod merchant_key_store;
 pub mod payment_address;
 pub mod payment_method_data;
+pub mod payment_methods;
 pub mod payments;
 #[cfg(feature = "payouts")]
 pub mod payouts;
diff --git a/crates/hyperswitch_domain_models/src/payment_methods.rs b/crates/hyperswitch_domain_models/src/payment_methods.rs
new file mode 100644
index 00000000000..ef8d90d0e43
--- /dev/null
+++ b/crates/hyperswitch_domain_models/src/payment_methods.rs
@@ -0,0 +1,380 @@
+use common_utils::{
+    crypto::OptionalEncryptableValue,
+    // date_time,
+    // encryption::Encryption,
+    errors::{CustomResult, ValidationError},
+    pii,
+    type_name,
+    types::keymanager,
+};
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use masking::{PeekInterface, Secret};
+use time::PrimitiveDateTime;
+
+use crate::type_encryption::{crypto_operation, AsyncLift, CryptoOperation};
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
+#[derive(Clone, Debug)]
+pub struct PaymentMethod {
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub payment_method_id: String,
+    pub accepted_currency: Option<Vec<storage_enums::Currency>>,
+    pub scheme: Option<String>,
+    pub token: Option<String>,
+    pub cardholder_name: Option<Secret<String>>,
+    pub issuer_name: Option<String>,
+    pub issuer_country: Option<String>,
+    pub payer_country: Option<Vec<String>>,
+    pub is_stored: Option<bool>,
+    pub swift_code: Option<String>,
+    pub direct_debit_token: Option<String>,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified: PrimitiveDateTime,
+    pub payment_method: Option<storage_enums::PaymentMethod>,
+    pub payment_method_type: Option<storage_enums::PaymentMethodType>,
+    pub payment_method_issuer: Option<String>,
+    pub payment_method_issuer_code: Option<storage_enums::PaymentMethodIssuerCode>,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub payment_method_data: OptionalEncryptableValue,
+    pub locker_id: Option<String>,
+    pub last_used_at: PrimitiveDateTime,
+    pub connector_mandate_details: Option<serde_json::Value>,
+    pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub status: storage_enums::PaymentMethodStatus,
+    pub network_transaction_id: Option<String>,
+    pub client_secret: Option<String>,
+    pub payment_method_billing_address: OptionalEncryptableValue,
+    pub updated_by: Option<String>,
+    pub version: common_enums::ApiVersion,
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+#[derive(Clone, Debug)]
+pub struct PaymentMethod {
+    pub customer_id: common_utils::id_type::CustomerId,
+    pub merchant_id: common_utils::id_type::MerchantId,
+    pub created_at: PrimitiveDateTime,
+    pub last_modified: PrimitiveDateTime,
+    pub payment_method: Option<storage_enums::PaymentMethod>,
+    pub payment_method_type: Option<storage_enums::PaymentMethodType>,
+    pub metadata: Option<pii::SecretSerdeValue>,
+    pub payment_method_data: OptionalEncryptableValue,
+    pub locker_id: Option<String>,
+    pub last_used_at: PrimitiveDateTime,
+    pub connector_mandate_details: Option<pii::SecretSerdeValue>,
+    pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub status: storage_enums::PaymentMethodStatus,
+    pub network_transaction_id: Option<String>,
+    pub client_secret: Option<String>,
+    pub payment_method_billing_address: OptionalEncryptableValue,
+    pub updated_by: Option<String>,
+    pub locker_fingerprint_id: Option<String>,
+    pub id: String,
+    pub version: common_enums::ApiVersion,
+}
+
+impl PaymentMethod {
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
+    pub fn get_id(&self) -> &String {
+        &self.payment_method_id
+    }
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    pub fn get_id(&self) -> &String {
+        &self.id
+    }
+}
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
+#[async_trait::async_trait]
+impl super::behaviour::Conversion for PaymentMethod {
+    type DstType = diesel_models::payment_method::PaymentMethod;
+    type NewDstType = diesel_models::payment_method::PaymentMethodNew;
+    async fn convert(self) -> CustomResult<Self::DstType, ValidationError> {
+        Ok(Self::DstType {
+            customer_id: self.customer_id,
+            merchant_id: self.merchant_id,
+            payment_method_id: self.payment_method_id,
+            accepted_currency: self.accepted_currency,
+            scheme: self.scheme,
+            token: self.token,
+            cardholder_name: self.cardholder_name,
+            issuer_name: self.issuer_name,
+            issuer_country: self.issuer_country,
+            payer_country: self.payer_country,
+            is_stored: self.is_stored,
+            swift_code: self.swift_code,
+            direct_debit_token: self.direct_debit_token,
+            created_at: self.created_at,
+            last_modified: self.last_modified,
+            payment_method: self.payment_method,
+            payment_method_type: self.payment_method_type,
+            payment_method_issuer: self.payment_method_issuer,
+            payment_method_issuer_code: self.payment_method_issuer_code,
+            metadata: self.metadata,
+            payment_method_data: self.payment_method_data.map(|val| val.into()),
+            locker_id: self.locker_id,
+            last_used_at: self.last_used_at,
+            connector_mandate_details: self.connector_mandate_details,
+            customer_acceptance: self.customer_acceptance,
+            status: self.status,
+            network_transaction_id: self.network_transaction_id,
+            client_secret: self.client_secret,
+            payment_method_billing_address: self
+                .payment_method_billing_address
+                .map(|val| val.into()),
+            updated_by: self.updated_by,
+            version: self.version,
+        })
+    }
+
+    async fn convert_back(
+        state: &keymanager::KeyManagerState,
+        item: Self::DstType,
+        key: &Secret<Vec<u8>>,
+        key_manager_identifier: keymanager::Identifier,
+    ) -> CustomResult<Self, ValidationError>
+    where
+        Self: Sized,
+    {
+        async {
+            Ok::<Self, error_stack::Report<common_utils::errors::CryptoError>>(Self {
+                customer_id: item.customer_id,
+                merchant_id: item.merchant_id,
+                payment_method_id: item.payment_method_id,
+                accepted_currency: item.accepted_currency,
+                scheme: item.scheme,
+                token: item.token,
+                cardholder_name: item.cardholder_name,
+                issuer_name: item.issuer_name,
+                issuer_country: item.issuer_country,
+                payer_country: item.payer_country,
+                is_stored: item.is_stored,
+                swift_code: item.swift_code,
+                direct_debit_token: item.direct_debit_token,
+                created_at: item.created_at,
+                last_modified: item.last_modified,
+                payment_method: item.payment_method,
+                payment_method_type: item.payment_method_type,
+                payment_method_issuer: item.payment_method_issuer,
+                payment_method_issuer_code: item.payment_method_issuer_code,
+                metadata: item.metadata,
+                payment_method_data: item
+                    .payment_method_data
+                    .async_lift(|inner| async {
+                        crypto_operation(
+                            state,
+                            type_name!(Self::DstType),
+                            CryptoOperation::DecryptOptional(inner),
+                            key_manager_identifier.clone(),
+                            key.peek(),
+                        )
+                        .await
+                        .and_then(|val| val.try_into_optionaloperation())
+                    })
+                    .await?,
+                locker_id: item.locker_id,
+                last_used_at: item.last_used_at,
+                connector_mandate_details: item.connector_mandate_details,
+                customer_acceptance: item.customer_acceptance,
+                status: item.status,
+                network_transaction_id: item.network_transaction_id,
+                client_secret: item.client_secret,
+                payment_method_billing_address: item
+                    .payment_method_billing_address
+                    .async_lift(|inner| async {
+                        crypto_operation(
+                            state,
+                            type_name!(Self::DstType),
+                            CryptoOperation::DecryptOptional(inner),
+                            key_manager_identifier.clone(),
+                            key.peek(),
+                        )
+                        .await
+                        .and_then(|val| val.try_into_optionaloperation())
+                    })
+                    .await?,
+                updated_by: item.updated_by,
+                version: item.version,
+            })
+        }
+        .await
+        .change_context(ValidationError::InvalidValue {
+            message: "Failed while decrypting payment method data".to_string(),
+        })
+    }
+
+    async fn construct_new(self) -> CustomResult<Self::NewDstType, ValidationError> {
+        Ok(Self::NewDstType {
+            customer_id: self.customer_id,
+            merchant_id: self.merchant_id,
+            payment_method_id: self.payment_method_id,
+            accepted_currency: self.accepted_currency,
+            scheme: self.scheme,
+            token: self.token,
+            cardholder_name: self.cardholder_name,
+            issuer_name: self.issuer_name,
+            issuer_country: self.issuer_country,
+            payer_country: self.payer_country,
+            is_stored: self.is_stored,
+            swift_code: self.swift_code,
+            direct_debit_token: self.direct_debit_token,
+            created_at: self.created_at,
+            last_modified: self.last_modified,
+            payment_method: self.payment_method,
+            payment_method_type: self.payment_method_type,
+            payment_method_issuer: self.payment_method_issuer,
+            payment_method_issuer_code: self.payment_method_issuer_code,
+            metadata: self.metadata,
+            payment_method_data: self.payment_method_data.map(|val| val.into()),
+            locker_id: self.locker_id,
+            last_used_at: self.last_used_at,
+            connector_mandate_details: self.connector_mandate_details,
+            customer_acceptance: self.customer_acceptance,
+            status: self.status,
+            network_transaction_id: self.network_transaction_id,
+            client_secret: self.client_secret,
+            payment_method_billing_address: self
+                .payment_method_billing_address
+                .map(|val| val.into()),
+            updated_by: self.updated_by,
+            version: self.version,
+        })
+    }
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+#[async_trait::async_trait]
+impl super::behaviour::Conversion for PaymentMethod {
+    type DstType = diesel_models::payment_method::PaymentMethod;
+    type NewDstType = diesel_models::payment_method::PaymentMethodNew;
+    async fn convert(self) -> CustomResult<Self::DstType, ValidationError> {
+        Ok(Self::DstType {
+            customer_id: self.customer_id,
+            merchant_id: self.merchant_id,
+            id: self.id,
+            created_at: self.created_at,
+            last_modified: self.last_modified,
+            payment_method: self.payment_method,
+            payment_method_type: self.payment_method_type,
+            metadata: self.metadata,
+            payment_method_data: self.payment_method_data.map(|val| val.into()),
+            locker_id: self.locker_id,
+            last_used_at: self.last_used_at,
+            connector_mandate_details: self.connector_mandate_details,
+            customer_acceptance: self.customer_acceptance,
+            status: self.status,
+            network_transaction_id: self.network_transaction_id,
+            client_secret: self.client_secret,
+            payment_method_billing_address: self
+                .payment_method_billing_address
+                .map(|val| val.into()),
+            updated_by: self.updated_by,
+            locker_fingerprint_id: self.locker_fingerprint_id,
+            version: self.version,
+        })
+    }
+
+    async fn convert_back(
+        state: &keymanager::KeyManagerState,
+        item: Self::DstType,
+        key: &Secret<Vec<u8>>,
+        key_manager_identifier: keymanager::Identifier,
+    ) -> CustomResult<Self, ValidationError>
+    where
+        Self: Sized,
+    {
+        async {
+            Ok::<Self, error_stack::Report<common_utils::errors::CryptoError>>(Self {
+                customer_id: item.customer_id,
+                merchant_id: item.merchant_id,
+                id: item.id,
+                created_at: item.created_at,
+                last_modified: item.last_modified,
+                payment_method: item.payment_method,
+                payment_method_type: item.payment_method_type,
+                metadata: item.metadata,
+                payment_method_data: item
+                    .payment_method_data
+                    .async_lift(|inner| async {
+                        crypto_operation(
+                            state,
+                            type_name!(Self::DstType),
+                            CryptoOperation::DecryptOptional(inner),
+                            key_manager_identifier.clone(),
+                            key.peek(),
+                        )
+                        .await
+                        .and_then(|val| val.try_into_optionaloperation())
+                    })
+                    .await?,
+                locker_id: item.locker_id,
+                last_used_at: item.last_used_at,
+                connector_mandate_details: item.connector_mandate_details,
+                customer_acceptance: item.customer_acceptance,
+                status: item.status,
+                network_transaction_id: item.network_transaction_id,
+                client_secret: item.client_secret,
+                payment_method_billing_address: item
+                    .payment_method_billing_address
+                    .async_lift(|inner| async {
+                        crypto_operation(
+                            state,
+                            type_name!(Self::DstType),
+                            CryptoOperation::DecryptOptional(inner),
+                            key_manager_identifier.clone(),
+                            key.peek(),
+                        )
+                        .await
+                        .and_then(|val| val.try_into_optionaloperation())
+                    })
+                    .await?,
+                updated_by: item.updated_by,
+                locker_fingerprint_id: item.locker_fingerprint_id,
+                version: item.version,
+            })
+        }
+        .await
+        .change_context(ValidationError::InvalidValue {
+            message: "Failed while decrypting payment method data".to_string(),
+        })
+    }
+
+    async fn construct_new(self) -> CustomResult<Self::NewDstType, ValidationError> {
+        Ok(Self::NewDstType {
+            customer_id: self.customer_id,
+            merchant_id: self.merchant_id,
+            id: self.id,
+            created_at: self.created_at,
+            last_modified: self.last_modified,
+            payment_method: self.payment_method,
+            payment_method_type: self.payment_method_type,
+            metadata: self.metadata,
+            payment_method_data: self.payment_method_data.map(|val| val.into()),
+            locker_id: self.locker_id,
+            last_used_at: self.last_used_at,
+            connector_mandate_details: self.connector_mandate_details,
+            customer_acceptance: self.customer_acceptance,
+            status: self.status,
+            network_transaction_id: self.network_transaction_id,
+            client_secret: self.client_secret,
+            payment_method_billing_address: self
+                .payment_method_billing_address
+                .map(|val| val.into()),
+            updated_by: self.updated_by,
+            locker_fingerprint_id: self.locker_fingerprint_id,
+            version: self.version,
+        })
+    }
+}
diff --git a/crates/router/src/core/customers.rs b/crates/router/src/core/customers.rs
index a38078b5a2b..8de8a42763a 100644
--- a/crates/router/src/core/customers.rs
+++ b/crates/router/src/core/customers.rs
@@ -532,7 +532,11 @@ pub async fn list_customers(
     Ok(services::ApplicationResponse::Json(customers))
 }
 
-#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "customer_v2"),
+    not(feature = "payment_methods_v2")
+))]
 #[instrument(skip_all)]
 pub async fn delete_customer(
     state: SessionState,
@@ -566,6 +570,8 @@ pub async fn delete_customer(
 
     match db
         .find_payment_method_by_customer_id_merchant_id_list(
+            key_manager_state,
+            &key_store,
             &req.customer_id,
             merchant_account.get_id(),
             None,
@@ -586,6 +592,8 @@ pub async fn delete_customer(
                     .switch()?;
                 }
                 db.delete_payment_method_by_merchant_id_payment_method_id(
+                    key_manager_state,
+                    &key_store,
                     merchant_account.get_id(),
                     &pm.payment_method_id,
                 )
diff --git a/crates/router/src/core/locker_migration.rs b/crates/router/src/core/locker_migration.rs
index 107586ee26d..8fd4b10ed3f 100644
--- a/crates/router/src/core/locker_migration.rs
+++ b/crates/router/src/core/locker_migration.rs
@@ -10,7 +10,6 @@ use common_utils::{errors::CustomResult, id_type};
     not(feature = "payment_methods_v2")
 ))]
 use diesel_models::enums as storage_enums;
-use diesel_models::PaymentMethod;
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
 use error_stack::FutureExt;
 use error_stack::ResultExt;
@@ -36,7 +35,11 @@ use crate::services::logger;
 use crate::types::api;
 use crate::{errors, routes::SessionState, services, types::domain};
 
-#[cfg(all(feature = "v2", feature = "customer_v2"))]
+#[cfg(all(
+    feature = "v2",
+    feature = "customer_v2",
+    feature = "payment_methods_v2"
+))]
 pub async fn rust_locker_migration(
     _state: SessionState,
     _merchant_id: &id_type::MerchantId,
@@ -44,7 +47,11 @@ pub async fn rust_locker_migration(
     todo!()
 }
 
-#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "customer_v2"),
+    not(feature = "payment_methods_v2")
+))]
 pub async fn rust_locker_migration(
     state: SessionState,
     merchant_id: &id_type::MerchantId,
@@ -86,6 +93,8 @@ pub async fn rust_locker_migration(
     for customer in domain_customers {
         let result = db
             .find_payment_method_by_customer_id_merchant_id_list(
+                key_manager_state,
+                &key_store,
                 &customer.customer_id,
                 merchant_id,
                 None,
@@ -122,7 +131,7 @@ pub async fn rust_locker_migration(
 ))]
 pub async fn call_to_locker(
     state: &SessionState,
-    payment_methods: Vec<PaymentMethod>,
+    payment_methods: Vec<domain::PaymentMethod>,
     customer_id: &id_type::CustomerId,
     merchant_id: &id_type::MerchantId,
     merchant_account: &domain::MerchantAccount,
@@ -220,7 +229,7 @@ pub async fn call_to_locker(
 #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
 pub async fn call_to_locker(
     _state: &SessionState,
-    _payment_methods: Vec<PaymentMethod>,
+    _payment_methods: Vec<domain::PaymentMethod>,
     _customer_id: &id_type::CustomerId,
     _merchant_id: &id_type::MerchantId,
     _merchant_account: &domain::MerchantAccount,
diff --git a/crates/router/src/core/mandate/helpers.rs b/crates/router/src/core/mandate/helpers.rs
index e5fa762aeb0..96f00a0d20e 100644
--- a/crates/router/src/core/mandate/helpers.rs
+++ b/crates/router/src/core/mandate/helpers.rs
@@ -90,5 +90,5 @@ pub struct MandateGenericData {
     pub recurring_mandate_payment_data:
         Option<hyperswitch_domain_models::router_data::RecurringMandatePaymentData>,
     pub mandate_connector: Option<payments::MandateConnectorDetails>,
-    pub payment_method_info: Option<diesel_models::PaymentMethod>,
+    pub payment_method_info: Option<domain::PaymentMethod>,
 }
diff --git a/crates/router/src/core/payment_methods.rs b/crates/router/src/core/payment_methods.rs
index fa030656528..88624cfd092 100644
--- a/crates/router/src/core/payment_methods.rs
+++ b/crates/router/src/core/payment_methods.rs
@@ -400,7 +400,7 @@ fn generate_task_id_for_payment_method_status_update_workflow(
 
 pub async fn add_payment_method_status_update_task(
     db: &dyn StorageInterface,
-    payment_method: &diesel_models::PaymentMethod,
+    payment_method: &domain::PaymentMethod,
     prev_status: enums::PaymentMethodStatus,
     curr_status: enums::PaymentMethodStatus,
     merchant_id: &common_utils::id_type::MerchantId,
@@ -410,7 +410,7 @@ pub async fn add_payment_method_status_update_task(
         created_at.saturating_add(Duration::seconds(consts::DEFAULT_SESSION_EXPIRY));
 
     let tracking_data = storage::PaymentMethodStatusTrackingData {
-        payment_method_id: payment_method.payment_method_id.clone(),
+        payment_method_id: payment_method.get_id().clone(),
         prev_status,
         curr_status,
         merchant_id: merchant_id.to_owned(),
@@ -421,7 +421,7 @@ pub async fn add_payment_method_status_update_task(
     let tag = [PAYMENT_METHOD_STATUS_TAG];
 
     let process_tracker_id = generate_task_id_for_payment_method_status_update_workflow(
-        payment_method.payment_method_id.as_str(),
+        payment_method.get_id().as_str(),
         &runner,
         task,
     );
@@ -443,7 +443,7 @@ pub async fn add_payment_method_status_update_task(
         .attach_printable_lazy(|| {
             format!(
                 "Failed while inserting PAYMENT_METHOD_STATUS_UPDATE reminder to process_tracker for payment_method_id: {}",
-                payment_method.payment_method_id.clone()
+                payment_method.get_id().clone()
             )
         })?;
 
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index f459210be7a..4fd3fc00089 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -26,7 +26,7 @@ use api_models::{
 use common_enums::{enums::MerchantStorageScheme, ConnectorType};
 use common_utils::{
     consts,
-    crypto::Encryptable,
+    crypto::{self, Encryptable},
     encryption::Encryption,
     ext_traits::{AsyncExt, Encode, StringExt, ValueExt},
     generate_id, id_type, type_name,
@@ -123,15 +123,15 @@ pub async fn create_payment_method(
     merchant_id: &id_type::MerchantId,
     pm_metadata: Option<serde_json::Value>,
     customer_acceptance: Option<serde_json::Value>,
-    payment_method_data: Option<Encryption>,
+    payment_method_data: crypto::OptionalEncryptableValue,
     key_store: &domain::MerchantKeyStore,
     connector_mandate_details: Option<serde_json::Value>,
     status: Option<enums::PaymentMethodStatus>,
     network_transaction_id: Option<String>,
     storage_scheme: MerchantStorageScheme,
-    payment_method_billing_address: Option<Encryption>,
+    payment_method_billing_address: crypto::OptionalEncryptableValue,
     card_scheme: Option<String>,
-) -> errors::CustomResult<storage::PaymentMethod, errors::ApiErrorResponse> {
+) -> errors::CustomResult<domain::PaymentMethod, errors::ApiErrorResponse> {
     let db = &*state.store;
     let customer = db
         .find_customer_by_customer_id_merchant_id(
@@ -153,7 +153,9 @@ pub async fn create_payment_method(
 
     let response = db
         .insert_payment_method(
-            storage::PaymentMethodNew {
+            &state.into(),
+            key_store,
+            domain::PaymentMethod {
                 customer_id: customer_id.to_owned(),
                 merchant_id: merchant_id.to_owned(),
                 payment_method_id: payment_method_id.to_string(),
@@ -184,6 +186,7 @@ pub async fn create_payment_method(
                 last_used_at: current_time,
                 payment_method_billing_address,
                 updated_by: None,
+                version: domain::consts::API_VERSION,
             },
             storage_scheme,
         )
@@ -230,7 +233,7 @@ pub async fn create_payment_method(
     _storage_scheme: MerchantStorageScheme,
     _payment_method_billing_address: Option<Encryption>,
     _card_scheme: Option<String>,
-) -> errors::CustomResult<storage::PaymentMethod, errors::ApiErrorResponse> {
+) -> errors::CustomResult<domain::PaymentMethod, errors::ApiErrorResponse> {
     todo!()
 }
 
@@ -279,7 +282,6 @@ pub fn store_default_payment_method(
 ) {
     todo!()
 }
-
 #[cfg(all(
     any(feature = "v1", feature = "v2"),
     not(feature = "payment_methods_v2")
@@ -292,13 +294,19 @@ pub async fn get_or_insert_payment_method(
     merchant_account: &domain::MerchantAccount,
     customer_id: &id_type::CustomerId,
     key_store: &domain::MerchantKeyStore,
-) -> errors::RouterResult<diesel_models::PaymentMethod> {
+) -> errors::RouterResult<domain::PaymentMethod> {
     let mut payment_method_id = resp.payment_method_id.clone();
     let mut locker_id = None;
     let db = &*state.store;
+    let key_manager_state = &(state.into());
     let payment_method = {
         let existing_pm_by_pmid = db
-            .find_payment_method(&payment_method_id, merchant_account.storage_scheme)
+            .find_payment_method(
+                key_manager_state,
+                key_store,
+                &payment_method_id,
+                merchant_account.storage_scheme,
+            )
             .await;
 
         if let Err(err) = existing_pm_by_pmid {
@@ -306,13 +314,15 @@ pub async fn get_or_insert_payment_method(
                 locker_id = Some(payment_method_id.clone());
                 let existing_pm_by_locker_id = db
                     .find_payment_method_by_locker_id(
+                        key_manager_state,
+                        key_store,
                         &payment_method_id,
                         merchant_account.storage_scheme,
                     )
                     .await;
 
                 match &existing_pm_by_locker_id {
-                    Ok(pm) => payment_method_id.clone_from(&pm.payment_method_id),
+                    Ok(pm) => payment_method_id.clone_from(pm.get_id()),
                     Err(_) => payment_method_id = generate_id(consts::ID_LENGTH, "pm"),
                 };
                 existing_pm_by_locker_id
@@ -363,7 +373,7 @@ pub async fn get_or_insert_payment_method(
     _merchant_account: &domain::MerchantAccount,
     _customer_id: &id_type::CustomerId,
     _key_store: &domain::MerchantKeyStore,
-) -> errors::RouterResult<diesel_models::PaymentMethod> {
+) -> errors::RouterResult<domain::PaymentMethod> {
     todo!()
 }
 
@@ -706,7 +716,9 @@ pub async fn skip_locker_call_and_migrate_payment_method(
 
     let response = db
         .insert_payment_method(
-            storage::PaymentMethodNew {
+            &(&state).into(),
+            key_store,
+            domain::PaymentMethod {
                 customer_id: customer_id.to_owned(),
                 merchant_id: merchant_id.to_owned(),
                 payment_method_id: payment_method_id.to_string(),
@@ -737,6 +749,7 @@ pub async fn skip_locker_call_and_migrate_payment_method(
                 last_used_at: current_time,
                 payment_method_billing_address: payment_method_billing_address.map(Into::into),
                 updated_by: None,
+                version: domain::consts::API_VERSION,
             },
             merchant_account.storage_scheme,
         )
@@ -895,7 +908,7 @@ pub async fn get_client_secret_or_add_payment_method(
 #[instrument(skip_all)]
 pub fn authenticate_pm_client_secret_and_check_expiry(
     req_client_secret: &String,
-    payment_method: &diesel_models::PaymentMethod,
+    payment_method: &domain::PaymentMethod,
 ) -> errors::CustomResult<bool, errors::ApiErrorResponse> {
     let stored_client_secret = payment_method
         .client_secret
@@ -945,7 +958,12 @@ pub async fn add_payment_method_data(
         .clone()
         .get_required_value("client_secret")?;
     let payment_method = db
-        .find_payment_method(pm_id.as_str(), merchant_account.storage_scheme)
+        .find_payment_method(
+            &((&state).into()),
+            &key_store,
+            pm_id.as_str(),
+            merchant_account.storage_scheme,
+        )
         .await
         .change_context(errors::ApiErrorResponse::PaymentMethodNotFound)
         .attach_printable("Unable to find payment method")?;
@@ -996,6 +1014,8 @@ pub async fn add_payment_method_data(
                         };
 
                         db.update_payment_method(
+                            &((&state).into()),
+                            &key_store,
                             payment_method,
                             pm_update,
                             merchant_account.storage_scheme,
@@ -1064,6 +1084,8 @@ pub async fn add_payment_method_data(
                         };
 
                         db.update_payment_method(
+                            &((&state).into()),
+                            &key_store,
                             payment_method,
                             pm_update,
                             merchant_account.storage_scheme,
@@ -1099,6 +1121,8 @@ pub async fn add_payment_method_data(
                     };
 
                     db.update_payment_method(
+                        &((&state).into()),
+                        &key_store,
                         payment_method,
                         pm_update,
                         merchant_account.storage_scheme,
@@ -1264,6 +1288,8 @@ pub async fn add_payment_method(
                     if let Err(err) = add_card_resp {
                         logger::error!(vault_err=?err);
                         db.delete_payment_method_by_merchant_id_payment_method_id(
+                            &(state.into()),
+                            key_store,
                             merchant_id,
                             &resp.payment_method_id,
                         )
@@ -1275,8 +1301,7 @@ pub async fn add_payment_method(
                     };
 
                     let existing_pm_data =
-                        get_card_details_without_locker_fallback(&existing_pm, state, key_store)
-                            .await?;
+                        get_card_details_without_locker_fallback(&existing_pm, state).await?;
 
                     let updated_card = Some(api::CardDetailFromLocker {
                         scheme: existing_pm.scheme.clone(),
@@ -1318,6 +1343,8 @@ pub async fn add_payment_method(
                     };
 
                     db.update_payment_method(
+                        &(state.into()),
+                        key_store,
                         existing_pm,
                         pm_update,
                         merchant_account.storage_scheme,
@@ -1394,14 +1421,14 @@ pub async fn insert_payment_method(
     connector_mandate_details: Option<serde_json::Value>,
     network_transaction_id: Option<String>,
     storage_scheme: MerchantStorageScheme,
-    payment_method_billing_address: Option<Encryption>,
-) -> errors::RouterResult<diesel_models::PaymentMethod> {
+    payment_method_billing_address: crypto::OptionalEncryptableValue,
+) -> errors::RouterResult<domain::PaymentMethod> {
     let pm_card_details = resp
         .card
         .clone()
         .map(|card| PaymentMethodsData::Card(CardDetailsPaymentMethod::from(card.clone())));
 
-    let pm_data_encrypted: Option<Encryptable<Secret<serde_json::Value>>> = pm_card_details
+    let pm_data_encrypted: crypto::OptionalEncryptableValue = pm_card_details
         .clone()
         .async_map(|pm_card| create_encrypted_data(state, key_store, pm_card))
         .await
@@ -1418,7 +1445,7 @@ pub async fn insert_payment_method(
         merchant_id,
         pm_metadata,
         customer_acceptance,
-        pm_data_encrypted.map(Into::into),
+        pm_data_encrypted,
         key_store,
         connector_mandate_details,
         None,
@@ -1449,7 +1476,7 @@ pub async fn insert_payment_method(
     network_transaction_id: Option<String>,
     storage_scheme: MerchantStorageScheme,
     payment_method_billing_address: Option<Encryption>,
-) -> errors::RouterResult<diesel_models::PaymentMethod> {
+) -> errors::RouterResult<domain::PaymentMethod> {
     let pm_card_details = match &resp.payment_method_data {
         Some(api::PaymentMethodResponseData::Card(card_data)) => Some(PaymentMethodsData::Card(
             CardDetailsPaymentMethod::from(card_data.clone()),
@@ -1503,7 +1530,12 @@ pub async fn update_customer_payment_method(
         let db = state.store.as_ref();
 
         let pm = db
-            .find_payment_method(payment_method_id, merchant_account.storage_scheme)
+            .find_payment_method(
+                &((&state).into()),
+                &key_store,
+                payment_method_id,
+                merchant_account.storage_scheme,
+            )
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
 
@@ -1528,36 +1560,28 @@ pub async fn update_customer_payment_method(
         }
 
         // Fetch the existing payment method data from db
-        let existing_card_data = domain::types::crypto_operation::<
-            serde_json::Value,
-            masking::WithType,
-        >(
-            &(&state).into(),
-            type_name!(payment_method::PaymentMethod),
-            domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-            Identifier::Merchant(key_store.merchant_id.clone()),
-            key_store.key.get_inner().peek(),
-        )
-        .await
-        .and_then(|val| val.try_into_optionaloperation())
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Failed to decrypt card details")?
-        .map(|x| x.into_inner().expose())
-        .map(
-            |value| -> Result<PaymentMethodsData, error_stack::Report<errors::ApiErrorResponse>> {
-                value
-                    .parse_value::<PaymentMethodsData>("PaymentMethodsData")
-                    .change_context(errors::ApiErrorResponse::InternalServerError)
-                    .attach_printable("Failed to deserialize payment methods data")
-            },
-        )
-        .transpose()?
-        .and_then(|pmd| match pmd {
-            PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
-            _ => None,
-        })
-        .ok_or(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("Failed to obtain decrypted card object from db")?;
+        let existing_card_data =
+            pm.payment_method_data
+                .clone()
+                .map(|x| x.into_inner().expose())
+                .map(
+                    |value| -> Result<
+                        PaymentMethodsData,
+                        error_stack::Report<errors::ApiErrorResponse>,
+                    > {
+                        value
+                            .parse_value::<PaymentMethodsData>("PaymentMethodsData")
+                            .change_context(errors::ApiErrorResponse::InternalServerError)
+                            .attach_printable("Failed to deserialize payment methods data")
+                    },
+                )
+                .transpose()?
+                .and_then(|pmd| match pmd {
+                    PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
+                    _ => None,
+                })
+                .ok_or(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable("Failed to obtain decrypted card object from db")?;
 
         let is_card_updation_required =
             validate_payment_method_update(card_update.clone(), existing_card_data.clone());
@@ -1675,10 +1699,16 @@ pub async fn update_customer_payment_method(
                 .payment_method_id
                 .clone_from(&pm.payment_method_id);
 
-            db.update_payment_method(pm, pm_update, merchant_account.storage_scheme)
-                .await
-                .change_context(errors::ApiErrorResponse::InternalServerError)
-                .attach_printable("Failed to update payment method in db")?;
+            db.update_payment_method(
+                &((&state).into()),
+                &key_store,
+                pm,
+                pm_update,
+                merchant_account.storage_scheme,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("Failed to update payment method in db")?;
 
             add_card_resp
         } else {
@@ -2165,9 +2195,15 @@ pub async fn call_to_locker_hs(
     Ok(stored_card)
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 pub async fn update_payment_method_metadata_and_last_used(
+    state: &routes::SessionState,
+    key_store: &domain::MerchantKeyStore,
     db: &dyn db::StorageInterface,
-    pm: payment_method::PaymentMethod,
+    pm: domain::PaymentMethod,
     pm_metadata: Option<serde_json::Value>,
     storage_scheme: MerchantStorageScheme,
 ) -> errors::CustomResult<(), errors::VaultError> {
@@ -2175,15 +2211,36 @@ pub async fn update_payment_method_metadata_and_last_used(
         metadata: pm_metadata,
         last_used_at: common_utils::date_time::now(),
     };
-    db.update_payment_method(pm, pm_update, storage_scheme)
+    db.update_payment_method(&(state.into()), key_store, pm, pm_update, storage_scheme)
+        .await
+        .change_context(errors::VaultError::UpdateInPaymentMethodDataTableFailed)?;
+    Ok(())
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+pub async fn update_payment_method_metadata_and_last_used(
+    state: &routes::SessionState,
+    key_store: &domain::MerchantKeyStore,
+    db: &dyn db::StorageInterface,
+    pm: domain::PaymentMethod,
+    pm_metadata: Option<serde_json::Value>,
+    storage_scheme: MerchantStorageScheme,
+) -> errors::CustomResult<(), errors::VaultError> {
+    let pm_update = payment_method::PaymentMethodUpdate::MetadataUpdateAndLastUsed {
+        metadata: pm_metadata.map(Secret::new),
+        last_used_at: common_utils::date_time::now(),
+    };
+    db.update_payment_method(&(state.into()), key_store, pm, pm_update, storage_scheme)
         .await
         .change_context(errors::VaultError::UpdateInPaymentMethodDataTableFailed)?;
     Ok(())
 }
 
 pub async fn update_payment_method_and_last_used(
+    state: &routes::SessionState,
+    key_store: &domain::MerchantKeyStore,
     db: &dyn db::StorageInterface,
-    pm: payment_method::PaymentMethod,
+    pm: domain::PaymentMethod,
     payment_method_update: Option<Encryption>,
     storage_scheme: MerchantStorageScheme,
 ) -> errors::CustomResult<(), errors::VaultError> {
@@ -2191,15 +2248,40 @@ pub async fn update_payment_method_and_last_used(
         payment_method_data: payment_method_update,
         last_used_at: common_utils::date_time::now(),
     };
-    db.update_payment_method(pm, pm_update, storage_scheme)
+    db.update_payment_method(&(state.into()), key_store, pm, pm_update, storage_scheme)
+        .await
+        .change_context(errors::VaultError::UpdateInPaymentMethodDataTableFailed)?;
+    Ok(())
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+pub async fn update_payment_method_connector_mandate_details(
+    state: &routes::SessionState,
+    key_store: &domain::MerchantKeyStore,
+    db: &dyn db::StorageInterface,
+    pm: domain::PaymentMethod,
+    connector_mandate_details: Option<serde_json::Value>,
+    storage_scheme: MerchantStorageScheme,
+) -> errors::CustomResult<(), errors::VaultError> {
+    let pm_update = payment_method::PaymentMethodUpdate::ConnectorMandateDetailsUpdate {
+        connector_mandate_details: connector_mandate_details.map(Secret::new),
+    };
+
+    db.update_payment_method(&(state.into()), key_store, pm, pm_update, storage_scheme)
         .await
         .change_context(errors::VaultError::UpdateInPaymentMethodDataTableFailed)?;
     Ok(())
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 pub async fn update_payment_method_connector_mandate_details(
+    state: &routes::SessionState,
+    key_store: &domain::MerchantKeyStore,
     db: &dyn db::StorageInterface,
-    pm: payment_method::PaymentMethod,
+    pm: domain::PaymentMethod,
     connector_mandate_details: Option<serde_json::Value>,
     storage_scheme: MerchantStorageScheme,
 ) -> errors::CustomResult<(), errors::VaultError> {
@@ -2207,7 +2289,7 @@ pub async fn update_payment_method_connector_mandate_details(
         connector_mandate_details,
     };
 
-    db.update_payment_method(pm, pm_update, storage_scheme)
+    db.update_payment_method(&(state.into()), key_store, pm, pm_update, storage_scheme)
         .await
         .change_context(errors::VaultError::UpdateInPaymentMethodDataTableFailed)?;
     Ok(())
@@ -2558,7 +2640,11 @@ fn get_val(str: String, val: &serde_json::Value) -> Option<String> {
         .and_then(|v| v.as_str())
         .map(|s| s.to_string())
 }
-#[cfg(all(feature = "v2", feature = "customer_v2"))]
+#[cfg(all(
+    feature = "v2",
+    feature = "customer_v2",
+    feature = "payment_methods_v2"
+))]
 pub async fn list_payment_methods(
     _state: routes::SessionState,
     _merchant_account: domain::MerchantAccount,
@@ -2568,7 +2654,11 @@ pub async fn list_payment_methods(
     todo!()
 }
 
-#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "customer_v2"),
+    not(feature = "payment_methods_v2")
+))]
 pub async fn list_payment_methods(
     state: routes::SessionState,
     merchant_account: domain::MerchantAccount,
@@ -2580,7 +2670,14 @@ pub async fn list_payment_methods(
     let key_manager_state = &(&state).into();
     let payment_intent = if let Some(cs) = &req.client_secret {
         if cs.starts_with("pm_") {
-            validate_payment_method_and_client_secret(cs, db, &merchant_account).await?;
+            validate_payment_method_and_client_secret(
+                &state,
+                cs,
+                db,
+                &merchant_account,
+                &key_store,
+            )
+            .await?;
             None
         } else {
             helpers::verify_payment_intent_time_and_client_secret(
@@ -2830,6 +2927,8 @@ pub async fn list_payment_methods(
             if wallet_pm_exists {
                 match db
                     .find_payment_method_by_customer_id_merchant_id_list(
+                        &((&state).into()),
+                        &key_store,
                        &customer.customer_id,
                        merchant_account.get_id(),
                         None,
@@ -3626,9 +3725,11 @@ fn should_collect_shipping_or_billing_details_from_wallet_connector(
 }
 
 async fn validate_payment_method_and_client_secret(
+    state: &routes::SessionState,
     cs: &String,
     db: &dyn db::StorageInterface,
     merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
 ) -> Result<(), error_stack::Report<errors::ApiErrorResponse>> {
     let pm_vec = cs.split("_secret").collect::<Vec<&str>>();
     let pm_id = pm_vec
@@ -3638,7 +3739,12 @@ async fn validate_payment_method_and_client_secret(
         })?;
 
     let payment_method = db
-        .find_payment_method(pm_id, merchant_account.storage_scheme)
+        .find_payment_method(
+            &(state.into()),
+            key_store,
+            pm_id,
+            merchant_account.storage_scheme,
+        )
         .await
         .change_context(errors::ApiErrorResponse::PaymentMethodNotFound)
         .attach_printable("Unable to find payment method")?;
@@ -4216,6 +4322,8 @@ pub async fn list_customer_payment_method(
 
     let resp = db
         .find_payment_method_by_customer_id_merchant_id_status(
+            &(state.into()),
+            &key_store,
             customer_id,
             merchant_account.get_id(),
             common_enums::PaymentMethodStatus::Active,
@@ -4275,23 +4383,22 @@ pub async fn list_customer_payment_method(
 
         // Retrieve the masked bank details to be sent as a response
         let bank_details = if payment_method == enums::PaymentMethod::BankDebit {
-            get_masked_bank_details(state, &pm, &key_store)
-                .await
-                .unwrap_or_else(|error| {
-                    logger::error!(?error);
-                    None
-                })
+            get_masked_bank_details(&pm).await.unwrap_or_else(|error| {
+                logger::error!(?error);
+                None
+            })
         } else {
             None
         };
 
-        let payment_method_billing = decrypt_generic_data::<api_models::payments::Address>(
-            state,
-            pm.payment_method_billing_address,
-            &key_store,
-        )
-        .await
-        .attach_printable("unable to decrypt payment method billing address details")?;
+        let payment_method_billing = pm
+            .payment_method_billing_address
+            .clone()
+            .map(|decrypted_data| decrypted_data.into_inner().expose())
+            .map(|decrypted_value| decrypted_value.parse_value("payment method billing address"))
+            .transpose()
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("unable to decrypt payment method billing address details")?;
         let connector_mandate_details = pm
             .connector_mandate_details
             .clone()
@@ -4416,15 +4523,16 @@ pub async fn list_customer_payment_method(
 async fn get_pm_list_context(
     state: &routes::SessionState,
     payment_method: &enums::PaymentMethod,
-    key_store: &domain::MerchantKeyStore,
-    pm: &diesel_models::PaymentMethod,
+    #[cfg(feature = "payouts")] key_store: &domain::MerchantKeyStore,
+    #[cfg(not(feature = "payouts"))] _key_store: &domain::MerchantKeyStore,
+    pm: &domain::PaymentMethod,
     #[cfg(feature = "payouts")] parent_payment_method_token: Option<String>,
     #[cfg(not(feature = "payouts"))] _parent_payment_method_token: Option<String>,
     is_payment_associated: bool,
 ) -> Result<Option<PaymentMethodListContext>, error_stack::Report<errors::ApiErrorResponse>> {
     let payment_method_retrieval_context = match payment_method {
         enums::PaymentMethod::Card => {
-            let card_details = get_card_details_with_locker_fallback(pm, state, key_store).await?;
+            let card_details = get_card_details_with_locker_fallback(pm, state).await?;
 
             card_details.as_ref().map(|card| PaymentMethodListContext {
                 card_details: Some(card.clone()),
@@ -4432,9 +4540,9 @@ async fn get_pm_list_context(
                 bank_transfer_details: None,
                 hyperswitch_token_data: is_payment_associated.then_some(
                     PaymentTokenData::permanent_card(
-                        Some(pm.payment_method_id.clone()),
-                        pm.locker_id.clone().or(Some(pm.payment_method_id.clone())),
-                        pm.locker_id.clone().unwrap_or(pm.payment_method_id.clone()),
+                        Some(pm.get_id().clone()),
+                        pm.locker_id.clone().or(Some(pm.get_id().clone())),
+                        pm.locker_id.clone().unwrap_or(pm.get_id().clone()),
                     ),
                 ),
             })
@@ -4442,7 +4550,7 @@ async fn get_pm_list_context(
 
         enums::PaymentMethod::BankDebit => {
             // Retrieve the pm_auth connector details so that it can be tokenized
-            let bank_account_token_data = get_bank_account_connector_details(state, pm, key_store)
+            let bank_account_token_data = get_bank_account_connector_details(pm)
                 .await
                 .unwrap_or_else(|err| {
                     logger::error!(error=?err);
@@ -4466,7 +4574,7 @@ async fn get_pm_list_context(
             #[cfg(feature = "payouts")]
             bank_transfer_details: None,
             hyperswitch_token_data: is_payment_associated
-                .then_some(PaymentTokenData::wallet_token(pm.payment_method_id.clone())),
+                .then_some(PaymentTokenData::wallet_token(pm.get_id().clone())),
         }),
 
         #[cfg(feature = "payouts")]
@@ -4479,7 +4587,7 @@ async fn get_pm_list_context(
                     parent_payment_method_token.as_ref(),
                     &pm.customer_id,
                     &pm.merchant_id,
-                    pm.locker_id.as_ref().unwrap_or(&pm.payment_method_id),
+                    pm.locker_id.as_ref().unwrap_or(pm.get_id()),
                 )
                 .await?,
             ),
@@ -4714,6 +4822,8 @@ pub async fn list_customer_payment_method(
 
     let saved_payment_methods = db
         .find_payment_method_by_customer_id_merchant_id_status(
+            key_manager_state,
+            &key_store,
             customer_id,
             merchant_account.get_id(),
             common_enums::PaymentMethodStatus::Active,
@@ -4794,7 +4904,7 @@ async fn generate_saved_pm_response(
     pm_list_context: (
         PaymentMethodListContext,
         Option<String>,
-        diesel_models::PaymentMethod,
+        domain::PaymentMethod,
     ),
     customer: &domain::Customer,
     payment_info: Option<&SavedPMLPaymentsInfo>,
@@ -4803,23 +4913,22 @@ async fn generate_saved_pm_response(
     let payment_method = pm.payment_method.get_required_value("payment_method")?;
 
     let bank_details = if payment_method == enums::PaymentMethod::BankDebit {
-        get_masked_bank_details(state, &pm, key_store)
-            .await
-            .unwrap_or_else(|err| {
-                logger::error!(error=?err);
-                None
-            })
+        get_masked_bank_details(&pm).await.unwrap_or_else(|err| {
+            logger::error!(error=?err);
+            None
+        })
     } else {
         None
     };
 
-    let payment_method_billing = decrypt_generic_data::<api_models::payments::Address>(
-        state,
-        pm.payment_method_billing_address,
-        key_store,
-    )
-    .await
-    .attach_printable("unable to decrypt payment method billing address details")?;
+    let payment_method_billing = pm
+        .payment_method_billing_address
+        .clone()
+        .map(|decrypted_data| decrypted_data.into_inner().expose())
+        .map(|decrypted_value| decrypted_value.parse_value("payment_method_billing_address"))
+        .transpose()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("unable to parse payment method billing address details")?;
 
     let connector_mandate_details = pm
         .connector_mandate_details
@@ -4875,12 +4984,12 @@ async fn generate_saved_pm_response(
 
     let pma = api::CustomerPaymentMethod {
         payment_token: parent_payment_method_token.clone(),
-        payment_method_id: pm.payment_method_id.clone(),
-        customer_id: pm.customer_id,
+        payment_method_id: pm.get_id().clone(),
+        customer_id: pm.customer_id.to_owned(),
         payment_method,
         payment_method_type: pm.payment_method_type,
         payment_method_data: pmd,
-        metadata: pm.metadata,
+        metadata: pm.metadata.clone(),
         recurring_enabled: mca_enabled,
         created: Some(pm.created_at),
         bank: bank_details,
@@ -4889,7 +4998,7 @@ async fn generate_saved_pm_response(
             && !(off_session_payment_flag && pm.connector_mandate_details.is_some()),
         last_used_at: Some(pm.last_used_at),
         is_default: customer.default_payment_method_id.is_some()
-            && customer.default_payment_method_id == Some(pm.payment_method_id),
+            && customer.default_payment_method_id.as_ref() == Some(pm.get_id()),
         billing: payment_method_billing,
     };
 
@@ -4983,31 +5092,18 @@ where
     not(feature = "payment_methods_v2")
 ))]
 pub async fn get_card_details_with_locker_fallback(
-    pm: &payment_method::PaymentMethod,
+    pm: &domain::PaymentMethod,
     state: &routes::SessionState,
-    key_store: &domain::MerchantKeyStore,
 ) -> errors::RouterResult<Option<api::CardDetailFromLocker>> {
-    let key = key_store.key.get_inner().peek();
-    let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let card_decrypted = domain::types::crypto_operation::<serde_json::Value, masking::WithType>(
-        &state.into(),
-        type_name!(payment_method::PaymentMethod),
-        domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-        identifier,
-        key,
-    )
-    .await
-    .and_then(|val| val.try_into_optionaloperation())
-    .change_context(errors::StorageError::DecryptionError)
-    .attach_printable("unable to decrypt card details")
-    .ok()
-    .flatten()
-    .map(|x| x.into_inner().expose())
-    .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
-    .and_then(|pmd| match pmd {
-        PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
-        _ => None,
-    });
+    let card_decrypted = pm
+        .payment_method_data
+        .clone()
+        .map(|x| x.into_inner().expose())
+        .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
+        .and_then(|pmd| match pmd {
+            PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
+            _ => None,
+        });
 
     Ok(if let Some(mut crd) = card_decrypted {
         crd.scheme.clone_from(&pm.scheme);
@@ -5022,31 +5118,18 @@ pub async fn get_card_details_with_locker_fallback(
 
 #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
 pub async fn get_card_details_with_locker_fallback(
-    pm: &payment_method::PaymentMethod,
+    pm: &domain::PaymentMethod,
     state: &routes::SessionState,
-    key_store: &domain::MerchantKeyStore,
 ) -> errors::RouterResult<Option<api::CardDetailFromLocker>> {
-    let key = key_store.key.get_inner().peek();
-    let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let card_decrypted = domain::types::crypto_operation::<serde_json::Value, masking::WithType>(
-        &state.into(),
-        type_name!(payment_method::PaymentMethod),
-        domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-        identifier,
-        key,
-    )
-    .await
-    .and_then(|val| val.try_into_optionaloperation())
-    .change_context(errors::StorageError::DecryptionError)
-    .attach_printable("unable to decrypt card details")
-    .ok()
-    .flatten()
-    .map(|x| x.into_inner().expose())
-    .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
-    .and_then(|pmd| match pmd {
-        PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
-        _ => None,
-    });
+    let card_decrypted = pm
+        .payment_method_data
+        .clone()
+        .map(|x| x.into_inner().expose())
+        .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
+        .and_then(|pmd| match pmd {
+            PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
+            _ => None,
+        });
 
     Ok(if let Some(crd) = card_decrypted {
         Some(crd)
@@ -5063,31 +5146,18 @@ pub async fn get_card_details_with_locker_fallback(
     not(feature = "payment_methods_v2")
 ))]
 pub async fn get_card_details_without_locker_fallback(
-    pm: &payment_method::PaymentMethod,
+    pm: &domain::PaymentMethod,
     state: &routes::SessionState,
-    key_store: &domain::MerchantKeyStore,
 ) -> errors::RouterResult<api::CardDetailFromLocker> {
-    let key = key_store.key.get_inner().peek();
-    let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let card_decrypted = domain::types::crypto_operation::<serde_json::Value, masking::WithType>(
-        &state.into(),
-        type_name!(payment_method::PaymentMethod),
-        domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-        identifier,
-        key,
-    )
-    .await
-    .and_then(|val| val.try_into_optionaloperation())
-    .change_context(errors::StorageError::DecryptionError)
-    .attach_printable("unable to decrypt card details")
-    .ok()
-    .flatten()
-    .map(|x| x.into_inner().expose())
-    .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
-    .and_then(|pmd| match pmd {
-        PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
-        _ => None,
-    });
+    let card_decrypted = pm
+        .payment_method_data
+        .clone()
+        .map(|x| x.into_inner().expose())
+        .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
+        .and_then(|pmd| match pmd {
+            PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
+            _ => None,
+        });
 
     Ok(if let Some(mut crd) = card_decrypted {
         crd.scheme.clone_from(&pm.scheme);
@@ -5102,31 +5172,18 @@ pub async fn get_card_details_without_locker_fallback(
 
 #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
 pub async fn get_card_details_without_locker_fallback(
-    pm: &payment_method::PaymentMethod,
+    pm: &domain::PaymentMethod,
     state: &routes::SessionState,
-    key_store: &domain::MerchantKeyStore,
 ) -> errors::RouterResult<api::CardDetailFromLocker> {
-    let key = key_store.key.get_inner().peek();
-    let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let card_decrypted = domain::types::crypto_operation::<serde_json::Value, masking::WithType>(
-        &state.into(),
-        type_name!(payment_method::PaymentMethod),
-        domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-        identifier,
-        key,
-    )
-    .await
-    .and_then(|val| val.try_into_optionaloperation())
-    .change_context(errors::StorageError::DecryptionError)
-    .attach_printable("unable to decrypt card details")
-    .ok()
-    .flatten()
-    .map(|x| x.into_inner().expose())
-    .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
-    .and_then(|pmd| match pmd {
-        PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
-        _ => None,
-    });
+    let card_decrypted = pm
+        .payment_method_data
+        .clone()
+        .map(|x| x.into_inner().expose())
+        .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
+        .and_then(|pmd| match pmd {
+            PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
+            _ => None,
+        });
     Ok(if let Some(crd) = card_decrypted {
         crd
     } else {
@@ -5139,13 +5196,13 @@ pub async fn get_card_details_without_locker_fallback(
 
 pub async fn get_card_details_from_locker(
     state: &routes::SessionState,
-    pm: &storage::PaymentMethod,
+    pm: &domain::PaymentMethod,
 ) -> errors::RouterResult<api::CardDetailFromLocker> {
     let card = get_card_from_locker(
         state,
         &pm.customer_id,
         &pm.merchant_id,
-        pm.locker_id.as_ref().unwrap_or(&pm.payment_method_id),
+        pm.locker_id.as_ref().unwrap_or(pm.get_id()),
     )
     .await
     .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -5159,7 +5216,7 @@ pub async fn get_card_details_from_locker(
 pub async fn get_lookup_key_from_locker(
     state: &routes::SessionState,
     payment_token: &str,
-    pm: &storage::PaymentMethod,
+    pm: &domain::PaymentMethod,
     merchant_key_store: &domain::MerchantKeyStore,
 ) -> errors::RouterResult<api::CardDetailFromLocker> {
     let card_detail = get_card_details_from_locker(state, pm).await?;
@@ -5177,25 +5234,11 @@ pub async fn get_lookup_key_from_locker(
 }
 
 async fn get_masked_bank_details(
-    state: &routes::SessionState,
-    pm: &payment_method::PaymentMethod,
-    key_store: &domain::MerchantKeyStore,
+    pm: &domain::PaymentMethod,
 ) -> errors::RouterResult<Option<MaskedBankDetails>> {
-    let key = key_store.key.get_inner().peek();
-    let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let payment_method_data =
-        domain::types::crypto_operation::<serde_json::Value, masking::WithType>(
-            &state.into(),
-            type_name!(payment_method::PaymentMethod),
-            domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-            identifier,
-            key,
-        )
-        .await
-        .and_then(|val| val.try_into_optionaloperation())
-        .change_context(errors::StorageError::DecryptionError)
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("unable to decrypt bank details")?
+    let payment_method_data = pm
+        .payment_method_data
+        .clone()
         .map(|x| x.into_inner().expose())
         .map(
             |v| -> Result<PaymentMethodsData, error_stack::Report<errors::ApiErrorResponse>> {
@@ -5220,25 +5263,11 @@ async fn get_masked_bank_details(
 }
 
 async fn get_bank_account_connector_details(
-    state: &routes::SessionState,
-    pm: &payment_method::PaymentMethod,
-    key_store: &domain::MerchantKeyStore,
+    pm: &domain::PaymentMethod,
 ) -> errors::RouterResult<Option<BankAccountTokenData>> {
-    let key = key_store.key.get_inner().peek();
-    let identifier = Identifier::Merchant(key_store.merchant_id.clone());
-    let payment_method_data =
-        domain::types::crypto_operation::<serde_json::Value, masking::WithType>(
-            &state.into(),
-            type_name!(payment_method::PaymentMethod),
-            domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-            identifier,
-            key,
-        )
-        .await
-        .and_then(|val| val.try_into_optionaloperation())
-        .change_context(errors::StorageError::DecryptionError)
-        .change_context(errors::ApiErrorResponse::InternalServerError)
-        .attach_printable("unable to decrypt bank details")?
+    let payment_method_data = pm
+        .payment_method_data
+        .clone()
         .map(|x| x.into_inner().expose())
         .map(
             |v| -> Result<PaymentMethodsData, error_stack::Report<errors::ApiErrorResponse>> {
@@ -5322,7 +5351,12 @@ pub async fn set_default_payment_method(
         .to_not_found_response(errors::ApiErrorResponse::CustomerNotFound)?;
     // check for the presence of payment_method
     let payment_method = db
-        .find_payment_method(&payment_method_id, storage_scheme)
+        .find_payment_method(
+            &(state.into()),
+            &key_store,
+            &payment_method_id,
+            storage_scheme,
+        )
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
     let pm = payment_method
@@ -5379,9 +5413,10 @@ pub async fn set_default_payment_method(
 }
 
 pub async fn update_last_used_at(
-    payment_method: &diesel_models::PaymentMethod,
+    payment_method: &domain::PaymentMethod,
     state: &routes::SessionState,
     storage_scheme: MerchantStorageScheme,
+    key_store: &domain::MerchantKeyStore,
 ) -> errors::RouterResult<()> {
     let update_last_used = storage::PaymentMethodUpdate::LastUsedUpdate {
         last_used_at: common_utils::date_time::now(),
@@ -5389,7 +5424,13 @@ pub async fn update_last_used_at(
 
     state
         .store
-        .update_payment_method(payment_method.clone(), update_last_used, storage_scheme)
+        .update_payment_method(
+            &(state.into()),
+            key_store,
+            payment_method.clone(),
+            update_last_used,
+            storage_scheme,
+        )
         .await
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Failed to update the last_used_at in db")?;
@@ -5456,7 +5497,7 @@ impl TempLockerCardSupport {
         state: &routes::SessionState,
         payment_token: &str,
         card: api::CardDetailFromLocker,
-        pm: &storage::PaymentMethod,
+        pm: &domain::PaymentMethod,
         merchant_key_store: &domain::MerchantKeyStore,
     ) -> errors::RouterResult<api::CardDetailFromLocker> {
         let card_number = card.card_number.clone().get_required_value("card_number")?;
@@ -5491,7 +5532,7 @@ impl TempLockerCardSupport {
             None,
             None,
             Some(pm.customer_id.clone()),
-            Some(pm.payment_method_id.to_string()),
+            Some(pm.get_id().to_string()),
         )
         .change_context(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("Error getting Value2 for locker")?;
@@ -5546,7 +5587,12 @@ pub async fn retrieve_payment_method(
 ) -> errors::RouterResponse<api::PaymentMethodResponse> {
     let db = state.store.as_ref();
     let pm = db
-        .find_payment_method(&pm.payment_method_id, merchant_account.storage_scheme)
+        .find_payment_method(
+            &((&state).into()),
+            &key_store,
+            &pm.payment_method_id,
+            merchant_account.storage_scheme,
+        )
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
 
@@ -5565,7 +5611,7 @@ pub async fn retrieve_payment_method(
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Failed while getting card details from locker")?
         } else {
-            get_card_details_without_locker_fallback(&pm, &state, &key_store).await?
+            get_card_details_without_locker_fallback(&pm, &state).await?
         };
         Some(card_detail)
     } else {
@@ -5602,7 +5648,12 @@ pub async fn retrieve_payment_method(
 ) -> errors::RouterResponse<api::PaymentMethodResponse> {
     let db = state.store.as_ref();
     let pm = db
-        .find_payment_method(&pm.payment_method_id, merchant_account.storage_scheme)
+        .find_payment_method(
+            &((&state).into()),
+            &key_store,
+            &pm.payment_method_id,
+            merchant_account.storage_scheme,
+        )
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
 
@@ -5612,7 +5663,7 @@ pub async fn retrieve_payment_method(
                 &state,
                 &pm.customer_id,
                 &pm.merchant_id,
-                pm.locker_id.as_ref().unwrap_or(&pm.payment_method_id),
+                pm.locker_id.as_ref().unwrap_or(pm.get_id()),
             )
             .await
             .change_context(errors::ApiErrorResponse::InternalServerError)
@@ -5621,7 +5672,7 @@ pub async fn retrieve_payment_method(
                 .change_context(errors::ApiErrorResponse::InternalServerError)
                 .attach_printable("Failed while getting card details from locker")?
         } else {
-            get_card_details_without_locker_fallback(&pm, &state, &key_store).await?
+            get_card_details_without_locker_fallback(&pm, &state).await?
         };
         Some(card_detail)
     } else {
@@ -5629,9 +5680,9 @@ pub async fn retrieve_payment_method(
     };
     Ok(services::ApplicationResponse::Json(
         api::PaymentMethodResponse {
-            merchant_id: pm.merchant_id,
-            customer_id: pm.customer_id,
-            payment_method_id: pm.payment_method_id,
+            merchant_id: pm.merchant_id.to_owned(),
+            customer_id: pm.customer_id.to_owned(),
+            payment_method_id: pm.get_id().clone(),
             payment_method: pm.payment_method,
             payment_method_type: pm.payment_method_type,
             metadata: pm.metadata,
@@ -5671,6 +5722,8 @@ pub async fn delete_payment_method(
     let key_manager_state = &(&state).into();
     let key = db
         .find_payment_method(
+            &((&state).into()),
+            &key_store,
             pm_id.payment_method_id.as_str(),
             merchant_account.storage_scheme,
         )
@@ -5707,6 +5760,8 @@ pub async fn delete_payment_method(
     }
 
     db.delete_payment_method_by_merchant_id_payment_method_id(
+        &((&state).into()),
+        &key_store,
         merchant_account.get_id(),
         pm_id.payment_method_id.as_str(),
     )
diff --git a/crates/router/src/core/payment_methods/transformers.rs b/crates/router/src/core/payment_methods/transformers.rs
index ed4f2d9d15f..da97fedf2cd 100644
--- a/crates/router/src/core/payment_methods/transformers.rs
+++ b/crates/router/src/core/payment_methods/transformers.rs
@@ -18,7 +18,7 @@ use crate::{
     headers,
     pii::{prelude::*, Secret},
     services::{api as services, encryption},
-    types::{api, storage},
+    types::{api, domain},
     utils::OptionExt,
 };
 
@@ -538,7 +538,7 @@ pub fn mk_delete_card_response(
     not(feature = "payment_methods_v2")
 ))]
 pub fn get_card_detail(
-    pm: &storage::PaymentMethod,
+    pm: &domain::PaymentMethod,
     response: Card,
 ) -> CustomResult<api::CardDetailFromLocker, errors::VaultError> {
     let card_number = response.card_number;
@@ -567,7 +567,7 @@ pub fn get_card_detail(
 
 #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
 pub fn get_card_detail(
-    pm: &storage::PaymentMethod,
+    _pm: &domain::PaymentMethod,
     response: Card,
 ) -> CustomResult<api::CardDetailFromLocker, errors::VaultError> {
     let card_number = response.card_number;
@@ -575,13 +575,7 @@ pub fn get_card_detail(
     //fetch form card bin
 
     let card_detail = api::CardDetailFromLocker {
-        issuer_country: pm
-            .issuer_country
-            .as_ref()
-            .map(|c| api_enums::CountryAlpha2::from_str(c))
-            .transpose()
-            .ok()
-            .flatten(),
+        issuer_country: None,
         last4_digits: Some(last4_digits),
         card_number: Some(card_number),
         expiry_month: Some(response.card_exp_month),
diff --git a/crates/router/src/core/payments.rs b/crates/router/src/core/payments.rs
index 678c9522a9b..8778079e728 100644
--- a/crates/router/src/core/payments.rs
+++ b/crates/router/src/core/payments.rs
@@ -2731,7 +2731,7 @@ where
     pub confirm: Option<bool>,
     pub force_sync: Option<bool>,
     pub payment_method_data: Option<domain::PaymentMethodData>,
-    pub payment_method_info: Option<storage::PaymentMethod>,
+    pub payment_method_info: Option<domain::PaymentMethod>,
     pub refunds: Vec<storage::Refund>,
     pub disputes: Vec<storage::Dispute>,
     pub attempts: Option<Vec<storage::PaymentAttempt>>,
@@ -3761,7 +3761,7 @@ pub async fn decide_multiplex_connector_for_normal_or_recurring_payment<F: Clone
                                             mandate_reference_record.connector_mandate_id.clone(),
                                         ),
                                         payment_method_id: Some(
-                                            payment_method_info.payment_method_id.clone(),
+                                            payment_method_info.get_id().clone(),
                                         ),
                                         update_history: None,
                                     },
@@ -3849,7 +3849,7 @@ pub fn is_network_transaction_id_flow(
     state: &SessionState,
     is_connector_agnostic_mit_enabled: Option<bool>,
     connector: enums::Connector,
-    payment_method_info: &storage::PaymentMethod,
+    payment_method_info: &domain::PaymentMethod,
 ) -> bool {
     let ntid_supported_connectors = &state
         .conf
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 38f0fd127f4..8ce183d3e3d 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -380,6 +380,23 @@ pub async fn get_address_by_id(
     }
 }
 
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+pub async fn get_token_pm_type_mandate_details(
+    _state: &SessionState,
+    _request: &api::PaymentsRequest,
+    _mandate_type: Option<api::MandateTransactionType>,
+    _merchant_account: &domain::MerchantAccount,
+    _merchant_key_store: &domain::MerchantKeyStore,
+    _payment_method_id: Option<String>,
+    _payment_intent_customer_id: Option<&id_type::CustomerId>,
+) -> RouterResult<MandateGenericData> {
+    todo!()
+}
+
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 pub async fn get_token_pm_type_mandate_details(
     state: &SessionState,
     request: &api::PaymentsRequest,
@@ -479,6 +496,8 @@ pub async fn get_token_pm_type_mandate_details(
                             let payment_method_info = state
                                 .store
                                 .find_payment_method(
+                                    &(state.into()),
+                                    merchant_key_store,
                                     payment_method_id,
                                     merchant_account.storage_scheme,
                                 )
@@ -542,6 +561,8 @@ pub async fn get_token_pm_type_mandate_details(
                             let customer_saved_pm_option = match state
                                 .store
                                 .find_payment_method_by_customer_id_merchant_id_list(
+                                    &(state.into()),
+                                    merchant_key_store,
                                     customer_id,
                                     merchant_account.get_id(),
                                     None,
@@ -596,6 +617,8 @@ pub async fn get_token_pm_type_mandate_details(
                                 state
                                     .store
                                     .find_payment_method(
+                                        &(state.into()),
+                                        merchant_key_store,
                                         &payment_method_id,
                                         merchant_account.storage_scheme,
                                     )
@@ -624,7 +647,12 @@ pub async fn get_token_pm_type_mandate_details(
                 .async_map(|payment_method_id| async move {
                     state
                         .store
-                        .find_payment_method(&payment_method_id, merchant_account.storage_scheme)
+                        .find_payment_method(
+                            &(state.into()),
+                            merchant_key_store,
+                            &payment_method_id,
+                            merchant_account.storage_scheme,
+                        )
                         .await
                         .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
                 })
@@ -716,7 +744,12 @@ pub async fn get_token_for_recurring_mandate(
     )?;
 
     let payment_method = db
-        .find_payment_method(payment_method_id.as_str(), merchant_account.storage_scheme)
+        .find_payment_method(
+            &(state.into()),
+            merchant_key_store,
+            payment_method_id.as_str(),
+            merchant_account.storage_scheme,
+        )
         .await
         .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
 
@@ -1900,15 +1933,21 @@ pub async fn retrieve_card_with_permanent_token(
 
 pub async fn retrieve_payment_method_from_db_with_token_data(
     state: &SessionState,
+    merchant_key_store: &domain::MerchantKeyStore,
     token_data: &storage::PaymentTokenData,
     storage_scheme: storage::enums::MerchantStorageScheme,
-) -> RouterResult<Option<storage::PaymentMethod>> {
+) -> RouterResult<Option<domain::PaymentMethod>> {
     match token_data {
         storage::PaymentTokenData::PermanentCard(data) => {
             if let Some(ref payment_method_id) = data.payment_method_id {
                 state
                     .store
-                    .find_payment_method(payment_method_id, storage_scheme)
+                    .find_payment_method(
+                        &(state.into()),
+                        merchant_key_store,
+                        payment_method_id,
+                        storage_scheme,
+                    )
                     .await
                     .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
                     .attach_printable("error retrieving payment method from DB")
@@ -1920,7 +1959,12 @@ pub async fn retrieve_payment_method_from_db_with_token_data(
 
         storage::PaymentTokenData::WalletToken(data) => state
             .store
-            .find_payment_method(&data.payment_method_id, storage_scheme)
+            .find_payment_method(
+                &(state.into()),
+                merchant_key_store,
+                &data.payment_method_id,
+                storage_scheme,
+            )
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)
             .attach_printable("error retrieveing payment method from DB")
@@ -2018,15 +2062,15 @@ pub async fn make_pm_data<'a, F: Clone, R>(
             if payment_method_info.payment_method == Some(storage_enums::PaymentMethod::Card) {
                 payment_data.token_data =
                     Some(storage::PaymentTokenData::PermanentCard(CardTokenData {
-                        payment_method_id: Some(payment_method_info.payment_method_id.clone()),
+                        payment_method_id: Some(payment_method_info.get_id().clone()),
                         locker_id: payment_method_info
                             .locker_id
                             .clone()
-                            .or(Some(payment_method_info.payment_method_id.clone())),
+                            .or(Some(payment_method_info.get_id().clone())),
                         token: payment_method_info
                             .locker_id
                             .clone()
-                            .unwrap_or(payment_method_info.payment_method_id.clone()),
+                            .unwrap_or(payment_method_info.get_id().clone()),
                     }));
             }
         }
diff --git a/crates/router/src/core/payments/operations/payment_confirm.rs b/crates/router/src/core/payments/operations/payment_confirm.rs
index 3e7944838a6..3edefd29466 100644
--- a/crates/router/src/core/payments/operations/payment_confirm.rs
+++ b/crates/router/src/core/payments/operations/payment_confirm.rs
@@ -11,8 +11,6 @@ use api_models::{
 use api_models::{payment_methods::PaymentMethodsData, payments::AdditionalPaymentData};
 use async_trait::async_trait;
 use common_utils::ext_traits::{AsyncExt, Encode, StringExt, ValueExt};
-#[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
-use common_utils::{type_name, types::keymanager::Identifier};
 use error_stack::{report, ResultExt};
 use futures::FutureExt;
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "customer_v2")))]
@@ -27,7 +25,6 @@ use super::{BoxedOperation, Domain, GetTracker, Operation, UpdateTracker, Valida
 use crate::{
     core::payment_methods::cards::create_encrypted_data,
     events::audit_events::{AuditEvent, AuditEventType},
-    types::domain::types::{crypto_operation, CryptoOperation},
 };
 use crate::{
     core::{
@@ -600,6 +597,7 @@ impl<F: Send + Clone> GetTracker<F, PaymentData<F>, api::PaymentsRequest> for Pa
 
             let payment_method_info = helpers::retrieve_payment_method_from_db_with_token_data(
                 state,
+                key_store,
                 &token_data,
                 storage_scheme,
             )
@@ -1160,24 +1158,11 @@ impl<F: Clone> UpdateTracker<F, PaymentData<F>, api::PaymentsRequest> for Paymen
             .transpose()
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to encode additional pm data")?;
-        let key_manager_state = &state.into();
+
         let encode_additional_pm_to_value = if let Some(ref pm) = payment_data.payment_method_info {
-            let key = key_store.key.get_inner().peek();
-
-            let card_detail_from_locker: Option<api::CardDetailFromLocker> =
-                crypto_operation::<serde_json::Value, masking::WithType>(
-                    key_manager_state,
-                    type_name!(storage::PaymentMethod),
-                    CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-                    Identifier::Merchant(key_store.merchant_id.clone()),
-                    key,
-                )
-                .await
-                .and_then(|val| val.try_into_optionaloperation())
-                .change_context(errors::StorageError::DecryptionError)
-                .attach_printable("unable to decrypt card details")
-                .ok()
-                .flatten()
+            let card_detail_from_locker: Option<api::CardDetailFromLocker> = pm
+                .payment_method_data
+                .clone()
                 .map(|x| x.into_inner().expose())
                 .and_then(|v| serde_json::from_value::<PaymentMethodsData>(v).ok())
                 .and_then(|pmd| match pmd {
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index a947d66bae5..6c0b748fc43 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -7,15 +7,13 @@ use api_models::{
 use async_trait::async_trait;
 use common_utils::{
     ext_traits::{AsyncExt, Encode, ValueExt},
-    type_name,
-    types::{keymanager::Identifier, MinorUnit},
+    types::MinorUnit,
 };
-use diesel_models::{ephemeral_key, PaymentMethod};
+use diesel_models::ephemeral_key;
 use error_stack::{self, ResultExt};
 use hyperswitch_domain_models::{
     mandates::{MandateData, MandateDetails},
     payments::{payment_attempt::PaymentAttempt, payment_intent::CustomerData},
-    type_encryption::{crypto_operation, CryptoOperation},
 };
 use masking::{ExposeInterface, PeekInterface, Secret};
 use router_derive::PaymentOperation;
@@ -877,8 +875,8 @@ impl PaymentCreate {
         browser_info: Option<serde_json::Value>,
         state: &SessionState,
         payment_method_billing_address_id: Option<String>,
-        payment_method_info: &Option<PaymentMethod>,
-        key_store: &domain::MerchantKeyStore,
+        payment_method_info: &Option<domain::PaymentMethod>,
+        _key_store: &domain::MerchantKeyStore,
         profile_id: common_utils::id_type::ProfileId,
         customer_acceptance: &Option<payments::CustomerAcceptance>,
     ) -> RouterResult<(
@@ -917,36 +915,28 @@ impl PaymentCreate {
             // If recurring payment is made using payment_method_id, then fetch payment_method_data from retrieved payment_method object
             additional_pm_data = payment_method_info
                 .as_ref()
-                .async_and_then(|pm_info| async {
-                    crypto_operation::<serde_json::Value, masking::WithType>(
-                        &state.into(),
-                        type_name!(PaymentMethod),
-                        CryptoOperation::DecryptOptional(pm_info.payment_method_data.clone()),
-                        Identifier::Merchant(key_store.merchant_id.clone()),
-                        key_store.key.get_inner().peek(),
-                    )
-                    .await
-                    .and_then(|val| val.try_into_optionaloperation())
-                    .map_err(|err| logger::error!("Failed to decrypt card details: {:?}", err))
-                    .ok()
-                    .flatten()
-                    .map(|x| x.into_inner().expose())
-                    .and_then(|v| {
-                        serde_json::from_value::<PaymentMethodsData>(v)
-                            .map_err(|err| {
-                                logger::error!(
-                                    "Unable to deserialize payment methods data: {:?}",
-                                    err
-                                )
-                            })
-                            .ok()
-                    })
-                    .and_then(|pmd| match pmd {
-                        PaymentMethodsData::Card(crd) => Some(api::CardDetailFromLocker::from(crd)),
-                        _ => None,
-                    })
+                .and_then(|pm_info| {
+                    pm_info
+                        .payment_method_data
+                        .clone()
+                        .map(|x| x.into_inner().expose())
+                        .and_then(|v| {
+                            serde_json::from_value::<PaymentMethodsData>(v)
+                                .map_err(|err| {
+                                    logger::error!(
+                                        "Unable to deserialize payment methods data: {:?}",
+                                        err
+                                    )
+                                })
+                                .ok()
+                        })
+                        .and_then(|pmd| match pmd {
+                            PaymentMethodsData::Card(crd) => {
+                                Some(api::CardDetailFromLocker::from(crd))
+                            }
+                            _ => None,
+                        })
                 })
-                .await
                 .map(|card| {
                     api_models::payments::AdditionalPaymentData::Card(Box::new(card.into()))
                 });
@@ -1032,7 +1022,7 @@ impl PaymentCreate {
                 offer_amount: None,
                 payment_method_id: payment_method_info
                     .as_ref()
-                    .map(|pm_info| pm_info.payment_method_id.clone()),
+                    .map(|pm_info| pm_info.get_id().clone()),
                 cancellation_reason: None,
                 error_code: None,
                 connector_metadata: None,
diff --git a/crates/router/src/core/payments/operations/payment_response.rs b/crates/router/src/core/payments/operations/payment_response.rs
index 5583d41a235..675c64aeb98 100644
--- a/crates/router/src/core/payments/operations/payment_response.rs
+++ b/crates/router/src/core/payments/operations/payment_response.rs
@@ -132,6 +132,7 @@ impl<F: Send + Clone> PostUpdateTracker<F, PaymentData<F>, types::PaymentsAuthor
                     payment_method_info,
                     state,
                     merchant_account.storage_scheme,
+                    key_store,
                 )
                 .await
                 .map_err(|e| {
@@ -193,8 +194,7 @@ impl<F: Send + Clone> PostUpdateTracker<F, PaymentData<F>, types::PaymentsAuthor
             Ok(())
         } else if should_avoid_saving {
             if let Some(pm_info) = &payment_data.payment_method_info {
-                payment_data.payment_attempt.payment_method_id =
-                    Some(pm_info.payment_method_id.clone());
+                payment_data.payment_attempt.payment_method_id = Some(pm_info.get_id().clone());
             };
             Ok(())
         } else {
@@ -436,7 +436,7 @@ impl<F: Clone> PostUpdateTracker<F, PaymentData<F>, types::PaymentsSyncData> for
         state: &SessionState,
         resp: &types::RouterData<F, types::PaymentsSyncData, types::PaymentsResponseData>,
         merchant_account: &domain::MerchantAccount,
-        _key_store: &domain::MerchantKeyStore,
+        key_store: &domain::MerchantKeyStore,
         payment_data: &mut PaymentData<F>,
         business_profile: &domain::BusinessProfile,
     ) -> CustomResult<(), errors::ApiErrorResponse>
@@ -445,6 +445,7 @@ impl<F: Clone> PostUpdateTracker<F, PaymentData<F>, types::PaymentsSyncData> for
     {
         update_payment_method_status_and_ntid(
             state,
+            key_store,
             payment_data,
             resp.status,
             resp.response.clone(),
@@ -751,7 +752,7 @@ impl<F: Clone> PostUpdateTracker<F, PaymentData<F>, types::CompleteAuthorizeData
         state: &SessionState,
         resp: &types::RouterData<F, types::CompleteAuthorizeData, types::PaymentsResponseData>,
         merchant_account: &domain::MerchantAccount,
-        _key_store: &domain::MerchantKeyStore,
+        key_store: &domain::MerchantKeyStore,
         payment_data: &mut PaymentData<F>,
         business_profile: &domain::BusinessProfile,
     ) -> CustomResult<(), errors::ApiErrorResponse>
@@ -760,6 +761,7 @@ impl<F: Clone> PostUpdateTracker<F, PaymentData<F>, types::CompleteAuthorizeData
     {
         update_payment_method_status_and_ntid(
             state,
+            key_store,
             payment_data,
             resp.status,
             resp.response.clone(),
@@ -1358,6 +1360,8 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                     connector_mandate_id,
                 )?;
             payment_methods::cards::update_payment_method_connector_mandate_details(
+                state,
+                key_store,
                 &*state.store,
                 payment_method.clone(),
                 connector_mandate_details,
@@ -1449,6 +1453,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
 
 async fn update_payment_method_status_and_ntid<F: Clone>(
     state: &SessionState,
+    key_store: &domain::MerchantKeyStore,
     payment_data: &mut PaymentData<F>,
     attempt_status: common_enums::AttemptStatus,
     payment_response: Result<types::PaymentsResponseData, ErrorResponse>,
@@ -1458,7 +1463,11 @@ async fn update_payment_method_status_and_ntid<F: Clone>(
     // If the payment_method is deleted then ignore the error related to retrieving payment method
     // This should be handled when the payment method is soft deleted
     if let Some(id) = &payment_data.payment_attempt.payment_method_id {
-        let payment_method = match state.store.find_payment_method(id, storage_scheme).await {
+        let payment_method = match state
+            .store
+            .find_payment_method(&(state.into()), key_store, id, storage_scheme)
+            .await
+        {
             Ok(payment_method) => payment_method,
             Err(error) => {
                 if error.current_context().is_db_not_found() {
@@ -1520,7 +1529,13 @@ async fn update_payment_method_status_and_ntid<F: Clone>(
 
         state
             .store
-            .update_payment_method(payment_method, pm_update, storage_scheme)
+            .update_payment_method(
+                &(state.into()),
+                key_store,
+                payment_method,
+                pm_update,
+                storage_scheme,
+            )
             .await
             .change_context(errors::ApiErrorResponse::InternalServerError)
             .attach_printable("Failed to update payment method in db")?;
diff --git a/crates/router/src/core/payments/operations/payment_status.rs b/crates/router/src/core/payments/operations/payment_status.rs
index f9aa1c2be56..85dab242de4 100644
--- a/crates/router/src/core/payments/operations/payment_status.rs
+++ b/crates/router/src/core/payments/operations/payment_status.rs
@@ -379,7 +379,12 @@ async fn get_tracker_for_sync<
     let payment_method_info =
         if let Some(ref payment_method_id) = payment_attempt.payment_method_id.clone() {
             match db
-                .find_payment_method(payment_method_id, storage_scheme)
+                .find_payment_method(
+                    &(state.into()),
+                    key_store,
+                    payment_method_id,
+                    storage_scheme,
+                )
                 .await
             {
                 Ok(payment_method) => Some(payment_method),
diff --git a/crates/router/src/core/payments/tokenization.rs b/crates/router/src/core/payments/tokenization.rs
index 6a922e1d407..609d5592ffd 100644
--- a/crates/router/src/core/payments/tokenization.rs
+++ b/crates/router/src/core/payments/tokenization.rs
@@ -247,6 +247,8 @@ where
                             let payment_method = {
                                 let existing_pm_by_pmid = db
                                     .find_payment_method(
+                                        &(state.into()),
+                                        key_store,
                                         &payment_method_id,
                                         merchant_account.storage_scheme,
                                     )
@@ -257,6 +259,8 @@ where
                                         locker_id = Some(payment_method_id.clone());
                                         let existing_pm_by_locker_id = db
                                             .find_payment_method_by_locker_id(
+                                                &(state.into()),
+                                                key_store,
                                                 &payment_method_id,
                                                 merchant_account.storage_scheme,
                                             )
@@ -289,6 +293,8 @@ where
                                         connector_token,
                                     )?;
                                     payment_methods::cards::update_payment_method_metadata_and_last_used(
+                                        state,
+                                        key_store,
                                         db,
                                         pm.clone(),
                                         pm_metadata,
@@ -308,7 +314,8 @@ where
                                                 connector_mandate_id.clone(),
                                             )?;
 
-                                        payment_methods::cards::update_payment_method_connector_mandate_details(db, pm, connector_mandate_details, merchant_account.storage_scheme).await.change_context(
+                                        payment_methods::cards::update_payment_method_connector_mandate_details(state,
+                                        key_store,db, pm, connector_mandate_details, merchant_account.storage_scheme).await.change_context(
                                         errors::ApiErrorResponse::InternalServerError,
                                     )
                                     .attach_printable("Failed to update payment method in db")?;
@@ -356,6 +363,8 @@ where
                                 let payment_method = {
                                     let existing_pm_by_pmid = db
                                         .find_payment_method(
+                                            &(state.into()),
+                                            key_store,
                                             &payment_method_id,
                                             merchant_account.storage_scheme,
                                         )
@@ -366,6 +375,8 @@ where
                                             locker_id = Some(payment_method_id.clone());
                                             let existing_pm_by_locker_id = db
                                                 .find_payment_method_by_locker_id(
+                                                    &(state.into()),
+                                                    key_store,
                                                     &payment_method_id,
                                                     merchant_account.storage_scheme,
                                                 )
@@ -406,7 +417,8 @@ where
                                                     connector_mandate_id.clone(),
                                                 )?;
 
-                                            payment_methods::cards::update_payment_method_connector_mandate_details(db, pm.clone(), connector_mandate_details, merchant_account.storage_scheme).await.change_context(
+                                            payment_methods::cards::update_payment_method_connector_mandate_details(  state,
+                                            key_store,db, pm.clone(), connector_mandate_details, merchant_account.storage_scheme).await.change_context(
                                             errors::ApiErrorResponse::InternalServerError,
                                         )
                                         .attach_printable("Failed to update payment method in db")?;
@@ -481,6 +493,8 @@ where
                                 if let Err(err) = add_card_resp {
                                     logger::error!(vault_err=?err);
                                     db.delete_payment_method_by_merchant_id_payment_method_id(
+                                        &(state.into()),
+                                        key_store,
                                         merchant_id,
                                         &resp.payment_method_id,
                                     )
@@ -495,9 +509,7 @@ where
                                         ))?
                                 };
 
-                                let existing_pm_data = payment_methods::cards::get_card_details_without_locker_fallback(&existing_pm,state,
-                                    key_store,
-                                )
+                                let existing_pm_data = payment_methods::cards::get_card_details_without_locker_fallback(&existing_pm,state)
                                 .await?;
 
                                 let updated_card = Some(CardDetailFromLocker {
@@ -539,6 +551,8 @@ where
                                     .attach_printable("Unable to encrypt payment method data")?;
 
                                 payment_methods::cards::update_payment_method_and_last_used(
+                                    state,
+                                    key_store,
                                     db,
                                     existing_pm,
                                     pm_data_encrypted.map(Into::into),
@@ -559,6 +573,8 @@ where
                             match state
                                 .store
                                 .find_payment_method_by_customer_id_merchant_id_list(
+                                    &(state.into()),
+                                    key_store,
                                     &customer_id,
                                     merchant_id,
                                     None,
@@ -594,6 +610,7 @@ where
                                 &customer_saved_pm,
                                 state,
                                 merchant_account.storage_scheme,
+                                key_store,
                             )
                             .await
                             .map_err(|e| {
@@ -1007,7 +1024,7 @@ pub fn add_connector_mandate_details_in_payment_method(
 }
 
 pub fn update_connector_mandate_details_in_payment_method(
-    payment_method: diesel_models::PaymentMethod,
+    payment_method: domain::PaymentMethod,
     payment_method_type: Option<storage_enums::PaymentMethodType>,
     authorized_amount: Option<i64>,
     authorized_currency: Option<storage_enums::Currency>,
diff --git a/crates/router/src/core/payments/transformers.rs b/crates/router/src/core/payments/transformers.rs
index b2070718892..a41205a496d 100644
--- a/crates/router/src/core/payments/transformers.rs
+++ b/crates/router/src/core/payments/transformers.rs
@@ -18,7 +18,6 @@ use crate::{
     connector::{Helcim, Nexinets},
     core::{
         errors::{self, RouterResponse, RouterResult},
-        payment_methods::cards::decrypt_generic_data,
         payments::{self, helpers},
         utils as core_utils,
     },
@@ -67,7 +66,7 @@ pub async fn construct_payment_router_data<'a, F, T>(
     payment_data: PaymentData<F>,
     connector_id: &str,
     merchant_account: &domain::MerchantAccount,
-    key_store: &domain::MerchantKeyStore,
+    _key_store: &domain::MerchantKeyStore,
     customer: &'a Option<domain::Customer>,
     merchant_connector_account: &helpers::MerchantConnectorAccountType,
     merchant_recipient_data: Option<types::MerchantRecipientData>,
@@ -158,22 +157,23 @@ where
         Some(merchant_connector_account),
     );
 
-    let unified_address =
-        if let Some(payment_method_info) = payment_data.payment_method_info.clone() {
-            let payment_method_billing = decrypt_generic_data::<Address>(
-                state,
-                payment_method_info.payment_method_billing_address,
-                key_store,
-            )
-            .await
-            .attach_printable("unable to decrypt payment method billing address details")?;
-            payment_data
-                .address
-                .clone()
-                .unify_with_payment_data_billing(payment_method_billing)
-        } else {
-            payment_data.address
-        };
+    let unified_address = if let Some(payment_method_info) =
+        payment_data.payment_method_info.clone()
+    {
+        let payment_method_billing = payment_method_info
+            .payment_method_billing_address
+            .map(|decrypted_data| decrypted_data.into_inner().expose())
+            .map(|decrypted_value| decrypted_value.parse_value("payment_method_billing_address"))
+            .transpose()
+            .change_context(errors::ApiErrorResponse::InternalServerError)
+            .attach_printable("unable to parse payment_method_billing_address")?;
+        payment_data
+            .address
+            .clone()
+            .unify_with_payment_data_billing(payment_method_billing)
+    } else {
+        payment_data.address
+    };
 
     crate::logger::debug!("unified address details {:?}", unified_address);
 
diff --git a/crates/router/src/core/payouts/helpers.rs b/crates/router/src/core/payouts/helpers.rs
index 1249b5d3050..7408a7ea2be 100644
--- a/crates/router/src/core/payouts/helpers.rs
+++ b/crates/router/src/core/payouts/helpers.rs
@@ -324,7 +324,12 @@ pub async fn save_payout_data_to_locker(
 
             // Use locker ref as payment_method_id
             let existing_pm_by_pmid = db
-                .find_payment_method(&locker_ref, merchant_account.storage_scheme)
+                .find_payment_method(
+                    &(state.into()),
+                    key_store,
+                    &locker_ref,
+                    merchant_account.storage_scheme,
+                )
                 .await;
 
             match existing_pm_by_pmid {
@@ -343,6 +348,8 @@ pub async fn save_payout_data_to_locker(
                     if err.current_context().is_db_not_found() {
                         match db
                             .find_payment_method_by_locker_id(
+                                &(state.into()),
+                                key_store,
                                 &locker_ref,
                                 merchant_account.storage_scheme,
                             )
@@ -570,6 +577,8 @@ pub async fn save_payout_data_to_locker(
         if let Err(err) = stored_resp {
             logger::error!(vault_err=?err);
             db.delete_payment_method_by_merchant_id_payment_method_id(
+                &(state.into()),
+                key_store,
                 merchant_account.get_id(),
                 &existing_pm.payment_method_id,
             )
@@ -585,10 +594,16 @@ pub async fn save_payout_data_to_locker(
         let pm_update = storage::PaymentMethodUpdate::PaymentMethodDataUpdate {
             payment_method_data: card_details_encrypted.map(Into::into),
         };
-        db.update_payment_method(existing_pm, pm_update, merchant_account.storage_scheme)
-            .await
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Failed to add payment method in db")?;
+        db.update_payment_method(
+            &(state.into()),
+            key_store,
+            existing_pm,
+            pm_update,
+            merchant_account.storage_scheme,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to add payment method in db")?;
     };
 
     // Store card_reference in payouts table
diff --git a/crates/router/src/core/pm_auth.rs b/crates/router/src/core/pm_auth.rs
index ba464d27f01..4380d60955b 100644
--- a/crates/router/src/core/pm_auth.rs
+++ b/crates/router/src/core/pm_auth.rs
@@ -12,9 +12,9 @@ pub mod transformers;
 use common_utils::{
     consts,
     crypto::{HmacSha256, SignMessage},
-    ext_traits::AsyncExt,
-    generate_id, type_name,
-    types::{self as util_types, keymanager::Identifier, AmountConvertor},
+    ext_traits::{AsyncExt, ValueExt},
+    generate_id,
+    types::{self as util_types, AmountConvertor},
 };
 use error_stack::ResultExt;
 use helpers::PaymentAuthConnectorDataExt;
@@ -42,12 +42,7 @@ use crate::{
     logger,
     routes::SessionState,
     services::{pm_auth as pm_auth_services, ApplicationResponse},
-    types::{
-        self,
-        domain::{self, types::crypto_operation},
-        storage,
-        transformers::ForeignTryFrom,
-    },
+    types::{self, domain, storage, transformers::ForeignTryFrom},
 };
 
 pub async fn create_link_token(
@@ -303,7 +298,6 @@ async fn store_bank_details_in_payment_methods(
     connector_details: (&str, Secret<String>),
     mca_id: common_utils::id_type::MerchantConnectorAccountId,
 ) -> RouterResult<()> {
-    let key = key_store.key.get_inner().peek();
     let db = &*state.clone().store;
     let (connector_name, access_token) = connector_details;
 
@@ -322,11 +316,31 @@ async fn store_bank_details_in_payment_methods(
         .customer_id
         .ok_or(ApiErrorResponse::CustomerNotFound)?;
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     let payment_methods = db
         .find_payment_method_by_customer_id_merchant_id_list(
+            &((&state).into()),
+            &key_store,
+            &customer_id,
+            merchant_account.get_id(),
+            None,
+        )
+        .await
+        .change_context(ApiErrorResponse::InternalServerError)?;
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    let payment_methods = db
+        .find_payment_method_by_customer_id_merchant_id_status(
+            &((&state).into()),
+            &key_store,
             &customer_id,
             merchant_account.get_id(),
+            common_enums::enums::PaymentMethodStatus::Active,
             None,
+            merchant_account.storage_scheme,
         )
         .await
         .change_context(ApiErrorResponse::InternalServerError)?;
@@ -334,7 +348,7 @@ async fn store_bank_details_in_payment_methods(
     let mut hash_to_payment_method: HashMap<
         String,
         (
-            storage::PaymentMethod,
+            domain::PaymentMethod,
             payment_methods::PaymentMethodDataBankCreds,
         ),
     > = HashMap::new();
@@ -343,33 +357,24 @@ async fn store_bank_details_in_payment_methods(
         if pm.payment_method == Some(enums::PaymentMethod::BankDebit)
             && pm.payment_method_data.is_some()
         {
-            let bank_details_pm_data = crypto_operation::<serde_json::Value, masking::WithType>(
-                &(&state).into(),
-                type_name!(storage::PaymentMethod),
-                domain::types::CryptoOperation::DecryptOptional(pm.payment_method_data.clone()),
-                Identifier::Merchant(key_store.merchant_id.clone()),
-                key,
-            )
-            .await
-            .and_then(|val| val.try_into_optionaloperation())
-            .change_context(ApiErrorResponse::InternalServerError)
-            .attach_printable("unable to decrypt bank account details")?
-            .map(|x| x.into_inner().expose())
-            .map(|v| {
-                serde_json::from_value::<payment_methods::PaymentMethodsData>(v)
-                    .change_context(errors::StorageError::DeserializationFailed)
-                    .attach_printable("Failed to deserialize Payment Method Auth config")
-            })
-            .transpose()
-            .unwrap_or_else(|error| {
-                logger::error!(?error);
-                None
-            })
-            .and_then(|pmd| match pmd {
-                payment_methods::PaymentMethodsData::BankDetails(bank_creds) => Some(bank_creds),
-                _ => None,
-            })
-            .ok_or(ApiErrorResponse::InternalServerError)?;
+            let bank_details_pm_data = pm
+                .payment_method_data
+                .clone()
+                .map(|x| x.into_inner().expose())
+                .map(|v| v.parse_value("PaymentMethodsData"))
+                .transpose()
+                .unwrap_or_else(|error| {
+                    logger::error!(?error);
+                    None
+                })
+                .and_then(|pmd| match pmd {
+                    payment_methods::PaymentMethodsData::BankDetails(bank_creds) => {
+                        Some(bank_creds)
+                    }
+                    _ => None,
+                })
+                .ok_or(ApiErrorResponse::InternalServerError)
+                .attach_printable("Unable to parse PaymentMethodsData")?;
 
             hash_to_payment_method.insert(
                 bank_details_pm_data.hash.clone(),
@@ -386,9 +391,8 @@ async fn store_bank_details_in_payment_methods(
         .clone()
         .expose();
 
-    let mut update_entries: Vec<(storage::PaymentMethod, storage::PaymentMethodUpdate)> =
-        Vec::new();
-    let mut new_entries: Vec<storage::PaymentMethodNew> = Vec::new();
+    let mut update_entries: Vec<(domain::PaymentMethod, storage::PaymentMethodUpdate)> = Vec::new();
+    let mut new_entries: Vec<domain::PaymentMethod> = Vec::new();
 
     for creds in bank_account_details_resp.credentials {
         let (account_number, hash_string) = match creds.account_details {
@@ -475,7 +479,11 @@ async fn store_bank_details_in_payment_methods(
                     .attach_printable("Unable to encrypt customer details")?;
             let pm_id = generate_id(consts::ID_LENGTH, "pm");
             let now = common_utils::date_time::now();
-            let pm_new = storage::PaymentMethodNew {
+            #[cfg(all(
+                any(feature = "v1", feature = "v2"),
+                not(feature = "payment_methods_v2")
+            ))]
+            let pm_new = domain::PaymentMethod {
                 customer_id: customer_id.clone(),
                 merchant_id: merchant_account.get_id().clone(),
                 payment_method_id: pm_id,
@@ -485,7 +493,7 @@ async fn store_bank_details_in_payment_methods(
                 payment_method_issuer: None,
                 scheme: None,
                 metadata: None,
-                payment_method_data: Some(encrypted_data.into()),
+                payment_method_data: Some(encrypted_data),
                 payment_method_issuer_code: None,
                 accepted_currency: None,
                 token: None,
@@ -506,6 +514,31 @@ async fn store_bank_details_in_payment_methods(
                 client_secret: None,
                 payment_method_billing_address: None,
                 updated_by: None,
+                version: domain::consts::API_VERSION,
+            };
+
+            #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+            let pm_new = domain::PaymentMethod {
+                customer_id: customer_id.clone(),
+                merchant_id: merchant_account.get_id().clone(),
+                id: pm_id,
+                payment_method: Some(enums::PaymentMethod::BankDebit),
+                payment_method_type: Some(creds.payment_method_type),
+                status: enums::PaymentMethodStatus::Active,
+                metadata: None,
+                payment_method_data: Some(encrypted_data.into()),
+                created_at: now,
+                last_modified: now,
+                locker_id: None,
+                last_used_at: now,
+                connector_mandate_details: None,
+                customer_acceptance: None,
+                network_transaction_id: None,
+                client_secret: None,
+                payment_method_billing_address: None,
+                updated_by: None,
+                locker_fingerprint_id: None,
+                version: domain::consts::API_VERSION,
             };
 
             new_entries.push(pm_new);
@@ -513,6 +546,8 @@ async fn store_bank_details_in_payment_methods(
     }
 
     store_in_db(
+        &state,
+        &key_store,
         update_entries,
         new_entries,
         db,
@@ -524,19 +559,26 @@ async fn store_bank_details_in_payment_methods(
 }
 
 async fn store_in_db(
-    update_entries: Vec<(storage::PaymentMethod, storage::PaymentMethodUpdate)>,
-    new_entries: Vec<storage::PaymentMethodNew>,
+    state: &SessionState,
+    key_store: &domain::MerchantKeyStore,
+    update_entries: Vec<(domain::PaymentMethod, storage::PaymentMethodUpdate)>,
+    new_entries: Vec<domain::PaymentMethod>,
     db: &dyn StorageInterface,
     storage_scheme: MerchantStorageScheme,
 ) -> RouterResult<()> {
+    let key_manager_state = &(state.into());
     let update_entries_futures = update_entries
         .into_iter()
-        .map(|(pm, pm_update)| db.update_payment_method(pm, pm_update, storage_scheme))
+        .map(|(pm, pm_update)| {
+            db.update_payment_method(key_manager_state, key_store, pm, pm_update, storage_scheme)
+        })
         .collect::<Vec<_>>();
 
     let new_entries_futures = new_entries
         .into_iter()
-        .map(|pm_new| db.insert_payment_method(pm_new, storage_scheme))
+        .map(|pm_new| {
+            db.insert_payment_method(key_manager_state, key_store, pm_new, storage_scheme)
+        })
         .collect::<Vec<_>>();
 
     let update_futures = futures::future::join_all(update_entries_futures);
diff --git a/crates/router/src/db/customers.rs b/crates/router/src/db/customers.rs
index 24402c343fa..3d8483f14f1 100644
--- a/crates/router/src/db/customers.rs
+++ b/crates/router/src/db/customers.rs
@@ -210,7 +210,7 @@ mod storage {
                 MerchantStorageScheme::RedisKv => {
                     let key = PartitionKey::MerchantIdCustomerId {
                         merchant_id,
-                        customer_id: customer_id.get_string_repr(),
+                        customer_id,
                     };
                     let field = format!("cust_{}", customer_id.get_string_repr());
                     Box::pin(db_utils::try_redis_get_else_try_database_get(
@@ -281,7 +281,7 @@ mod storage {
                 MerchantStorageScheme::RedisKv => {
                     let key = PartitionKey::MerchantIdCustomerId {
                         merchant_id,
-                        customer_id: customer_id.get_string_repr(),
+                        customer_id,
                     };
                     let field = format!("cust_{}", customer_id.get_string_repr());
                     Box::pin(db_utils::try_redis_get_else_try_database_get(
@@ -344,9 +344,9 @@ mod storage {
             let maybe_customer = match storage_scheme {
                 MerchantStorageScheme::PostgresOnly => database_call().await,
                 MerchantStorageScheme::RedisKv => {
-                    let key = PartitionKey::MerchantIdCustomerId {
+                    let key = PartitionKey::MerchantIdMerchantReferenceId {
                         merchant_id,
-                        customer_id: merchant_reference_id.get_string_repr(),
+                        merchant_reference_id: merchant_reference_id.get_string_repr(),
                     };
                     let field = format!("cust_{}", merchant_reference_id.get_string_repr());
                     Box::pin(db_utils::try_redis_get_else_try_database_get(
@@ -416,7 +416,7 @@ mod storage {
             };
             let key = PartitionKey::MerchantIdCustomerId {
                 merchant_id: &merchant_id,
-                customer_id: customer_id.get_string_repr(),
+                customer_id: &customer_id,
             };
             let field = format!("cust_{}", customer_id.get_string_repr());
             let storage_scheme = decide_storage_scheme::<_, diesel_models::Customer>(
@@ -564,7 +564,7 @@ mod storage {
                 MerchantStorageScheme::RedisKv => {
                     let key = PartitionKey::MerchantIdCustomerId {
                         merchant_id,
-                        customer_id: customer_id.get_string_repr(),
+                        customer_id,
                     };
                     let field = format!("cust_{}", customer_id.get_string_repr());
                     Box::pin(db_utils::try_redis_get_else_try_database_get(
@@ -747,7 +747,7 @@ mod storage {
                 MerchantStorageScheme::RedisKv => {
                     let key = PartitionKey::MerchantIdCustomerId {
                         merchant_id: &merchant_id,
-                        customer_id: customer_id.get_string_repr(),
+                        customer_id: &customer_id,
                     };
                     let field = format!("cust_{}", customer_id.get_string_repr());
 
diff --git a/crates/router/src/db/kafka_store.rs b/crates/router/src/db/kafka_store.rs
index 0293dd2b328..577ccfebabf 100644
--- a/crates/router/src/db/kafka_store.rs
+++ b/crates/router/src/db/kafka_store.rs
@@ -1693,35 +1693,53 @@ impl PaymentIntentInterface for KafkaStore {
 impl PaymentMethodInterface for KafkaStore {
     async fn find_payment_method(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         payment_method_id: &str,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         self.diesel_store
-            .find_payment_method(payment_method_id, storage_scheme)
+            .find_payment_method(state, key_store, payment_method_id, storage_scheme)
             .await
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn find_payment_method_by_customer_id_merchant_id_list(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         customer_id: &id_type::CustomerId,
         merchant_id: &id_type::MerchantId,
         limit: Option<i64>,
-    ) -> CustomResult<Vec<storage::PaymentMethod>, errors::StorageError> {
+    ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
         self.diesel_store
-            .find_payment_method_by_customer_id_merchant_id_list(customer_id, merchant_id, limit)
+            .find_payment_method_by_customer_id_merchant_id_list(
+                state,
+                key_store,
+                customer_id,
+                merchant_id,
+                limit,
+            )
             .await
     }
 
     async fn find_payment_method_by_customer_id_merchant_id_status(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         customer_id: &id_type::CustomerId,
         merchant_id: &id_type::MerchantId,
         status: common_enums::PaymentMethodStatus,
         limit: Option<i64>,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<Vec<storage::PaymentMethod>, errors::StorageError> {
+    ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
         self.diesel_store
             .find_payment_method_by_customer_id_merchant_id_status(
+                state,
+                key_store,
                 customer_id,
                 merchant_id,
                 status,
@@ -1731,6 +1749,10 @@ impl PaymentMethodInterface for KafkaStore {
             .await
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn get_payment_method_count_by_customer_id_merchant_id_status(
         &self,
         customer_id: &id_type::CustomerId,
@@ -1746,44 +1768,95 @@ impl PaymentMethodInterface for KafkaStore {
             .await
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn find_payment_method_by_locker_id(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         locker_id: &str,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         self.diesel_store
-            .find_payment_method_by_locker_id(locker_id, storage_scheme)
+            .find_payment_method_by_locker_id(state, key_store, locker_id, storage_scheme)
             .await
     }
 
     async fn insert_payment_method(
         &self,
-        m: storage::PaymentMethodNew,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        m: domain::PaymentMethod,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         self.diesel_store
-            .insert_payment_method(m, storage_scheme)
+            .insert_payment_method(state, key_store, m, storage_scheme)
             .await
     }
 
     async fn update_payment_method(
         &self,
-        payment_method: storage::PaymentMethod,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
         payment_method_update: storage::PaymentMethodUpdate,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         self.diesel_store
-            .update_payment_method(payment_method, payment_method_update, storage_scheme)
+            .update_payment_method(
+                state,
+                key_store,
+                payment_method,
+                payment_method_update,
+                storage_scheme,
+            )
             .await
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn delete_payment_method_by_merchant_id_payment_method_id(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         merchant_id: &id_type::MerchantId,
         payment_method_id: &str,
-    ) -> CustomResult<storage::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+        self.diesel_store
+            .delete_payment_method_by_merchant_id_payment_method_id(
+                state,
+                key_store,
+                merchant_id,
+                payment_method_id,
+            )
+            .await
+    }
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    async fn delete_payment_method(
+        &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+        self.diesel_store
+            .delete_payment_method(state, key_store, payment_method)
+            .await
+    }
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    async fn find_payment_method_by_fingerprint_id(
+        &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        fingerprint_id: &str,
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         self.diesel_store
-            .delete_payment_method_by_merchant_id_payment_method_id(merchant_id, payment_method_id)
+            .find_payment_method_by_fingerprint_id(state, key_store, fingerprint_id)
             .await
     }
 }
diff --git a/crates/router/src/db/payment_method.rs b/crates/router/src/db/payment_method.rs
index e182d62c3f3..62c34e7103b 100644
--- a/crates/router/src/db/payment_method.rs
+++ b/crates/router/src/db/payment_method.rs
@@ -1,43 +1,68 @@
-use common_utils::id_type;
+use common_utils::{id_type, types::keymanager::KeyManagerState};
 use diesel_models::payment_method::PaymentMethodUpdateInternal;
 use error_stack::ResultExt;
+use hyperswitch_domain_models::behaviour::{Conversion, ReverseConversion};
 
 use super::MockDb;
 use crate::{
     core::errors::{self, CustomResult},
-    types::storage::{self as storage_types, enums::MerchantStorageScheme},
+    types::{
+        domain,
+        storage::{self as storage_types, enums::MerchantStorageScheme},
+    },
 };
 
 #[async_trait::async_trait]
 pub trait PaymentMethodInterface {
     async fn find_payment_method(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         payment_method_id: &str,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError>;
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError>;
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn find_payment_method_by_locker_id(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         locker_id: &str,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError>;
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError>;
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn find_payment_method_by_customer_id_merchant_id_list(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         customer_id: &id_type::CustomerId,
         merchant_id: &id_type::MerchantId,
         limit: Option<i64>,
-    ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError>;
+    ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError>;
 
+    #[allow(clippy::too_many_arguments)]
     async fn find_payment_method_by_customer_id_merchant_id_status(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         customer_id: &id_type::CustomerId,
         merchant_id: &id_type::MerchantId,
         status: common_enums::PaymentMethodStatus,
         limit: Option<i64>,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError>;
+    ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError>;
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn get_payment_method_count_by_customer_id_merchant_id_status(
         &self,
         customer_id: &id_type::CustomerId,
@@ -47,29 +72,58 @@ pub trait PaymentMethodInterface {
 
     async fn insert_payment_method(
         &self,
-        payment_method_new: storage_types::PaymentMethodNew,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError>;
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError>;
 
     async fn update_payment_method(
         &self,
-        payment_method: storage_types::PaymentMethod,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
         payment_method_update: storage_types::PaymentMethodUpdate,
         storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError>;
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError>;
 
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    async fn delete_payment_method(
+        &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError>;
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    async fn find_payment_method_by_fingerprint_id(
+        &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        fingerprint_id: &str,
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError>;
+
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn delete_payment_method_by_merchant_id_payment_method_id(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         merchant_id: &id_type::MerchantId,
         payment_method_id: &str,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError>;
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError>;
 }
 
 #[cfg(feature = "kv_store")]
 mod storage {
-    use common_utils::{fallback_reverse_lookup_not_found, id_type};
+    use common_utils::{
+        fallback_reverse_lookup_not_found, id_type, types::keymanager::KeyManagerState,
+    };
     use diesel_models::{kv, PaymentMethodUpdateInternal};
     use error_stack::{report, ResultExt};
+    use hyperswitch_domain_models::behaviour::{Conversion, ReverseConversion};
     use redis_interface::HsetnxReply;
     use router_env::{instrument, tracing};
     use storage_impl::redis::kv_store::{
@@ -82,18 +136,27 @@ mod storage {
         core::errors::{self, utils::RedisErrorExt, CustomResult},
         db::reverse_lookup::ReverseLookupInterface,
         services::Store,
-        types::storage::{self as storage_types, enums::MerchantStorageScheme},
+        types::{
+            domain,
+            storage::{self as storage_types, enums::MerchantStorageScheme},
+        },
         utils::db_utils,
     };
 
     #[async_trait::async_trait]
     impl PaymentMethodInterface for Store {
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn find_payment_method(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             payment_method_id: &str,
             storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
             let database_call = || async {
                 storage_types::PaymentMethod::find_by_payment_method_id(&conn, payment_method_id)
@@ -106,43 +169,129 @@ mod storage {
                 Op::Find,
             )
             .await;
-            match storage_scheme {
-                MerchantStorageScheme::PostgresOnly => database_call().await,
-                MerchantStorageScheme::RedisKv => {
-                    let lookup_id = format!("payment_method_{}", payment_method_id);
-                    let lookup = fallback_reverse_lookup_not_found!(
-                        self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
-                            .await,
-                        database_call().await
-                    );
-
-                    let key = PartitionKey::CombinationKey {
-                        combination: &lookup.pk_id,
-                    };
+            let get_pm = || async {
+                match storage_scheme {
+                    MerchantStorageScheme::PostgresOnly => database_call().await,
+                    MerchantStorageScheme::RedisKv => {
+                        let lookup_id = format!("payment_method_{}", payment_method_id);
+                        let lookup = fallback_reverse_lookup_not_found!(
+                            self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
+                                .await,
+                            database_call().await
+                        );
 
-                    Box::pin(db_utils::try_redis_get_else_try_database_get(
-                        async {
-                            kv_wrapper(
-                                self,
-                                KvOperation::<storage_types::PaymentMethod>::HGet(&lookup.sk_id),
-                                key,
-                            )
-                            .await?
-                            .try_into_hget()
-                        },
-                        database_call,
-                    ))
+                        let key = PartitionKey::CombinationKey {
+                            combination: &lookup.pk_id,
+                        };
+
+                        Box::pin(db_utils::try_redis_get_else_try_database_get(
+                            async {
+                                kv_wrapper(
+                                    self,
+                                    KvOperation::<storage_types::PaymentMethod>::HGet(
+                                        &lookup.sk_id,
+                                    ),
+                                    key,
+                                )
+                                .await?
+                                .try_into_hget()
+                            },
+                            database_call,
+                        ))
+                        .await
+                    }
+                }
+            };
+
+            get_pm()
+                .await?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }
+
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        #[instrument(skip_all)]
+        async fn find_payment_method(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method_id: &str,
+            storage_scheme: MerchantStorageScheme,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            let database_call = || async {
+                storage_types::PaymentMethod::find_by_id(&conn, payment_method_id)
                     .await
+                    .map_err(|error| report!(errors::StorageError::from(error)))
+            };
+            let storage_scheme = decide_storage_scheme::<_, storage_types::PaymentMethod>(
+                self,
+                storage_scheme,
+                Op::Find,
+            )
+            .await;
+            let get_pm = || async {
+                match storage_scheme {
+                    MerchantStorageScheme::PostgresOnly => database_call().await,
+                    MerchantStorageScheme::RedisKv => {
+                        let lookup_id = format!("payment_method_{}", payment_method_id);
+                        let lookup = fallback_reverse_lookup_not_found!(
+                            self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
+                                .await,
+                            database_call().await
+                        );
+
+                        let key = PartitionKey::CombinationKey {
+                            combination: &lookup.pk_id,
+                        };
+
+                        Box::pin(db_utils::try_redis_get_else_try_database_get(
+                            async {
+                                kv_wrapper(
+                                    self,
+                                    KvOperation::<storage_types::PaymentMethod>::HGet(
+                                        &lookup.sk_id,
+                                    ),
+                                    key,
+                                )
+                                .await?
+                                .try_into_hget()
+                            },
+                            database_call,
+                        ))
+                        .await
+                    }
                 }
-            }
+            };
+
+            get_pm()
+                .await?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
 
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn find_payment_method_by_locker_id(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             locker_id: &str,
             storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
             let database_call = || async {
                 storage_types::PaymentMethod::find_by_locker_id(&conn, locker_id)
@@ -155,37 +304,56 @@ mod storage {
                 Op::Find,
             )
             .await;
-            match storage_scheme {
-                MerchantStorageScheme::PostgresOnly => database_call().await,
-                MerchantStorageScheme::RedisKv => {
-                    let lookup_id = format!("payment_method_locker_{}", locker_id);
-                    let lookup = fallback_reverse_lookup_not_found!(
-                        self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
-                            .await,
-                        database_call().await
-                    );
-
-                    let key = PartitionKey::CombinationKey {
-                        combination: &lookup.pk_id,
-                    };
+            let get_pm = || async {
+                match storage_scheme {
+                    MerchantStorageScheme::PostgresOnly => database_call().await,
+                    MerchantStorageScheme::RedisKv => {
+                        let lookup_id = format!("payment_method_locker_{}", locker_id);
+                        let lookup = fallback_reverse_lookup_not_found!(
+                            self.get_lookup_by_lookup_id(&lookup_id, storage_scheme)
+                                .await,
+                            database_call().await
+                        );
 
-                    Box::pin(db_utils::try_redis_get_else_try_database_get(
-                        async {
-                            kv_wrapper(
-                                self,
-                                KvOperation::<storage_types::PaymentMethod>::HGet(&lookup.sk_id),
-                                key,
-                            )
-                            .await?
-                            .try_into_hget()
-                        },
-                        database_call,
-                    ))
-                    .await
+                        let key = PartitionKey::CombinationKey {
+                            combination: &lookup.pk_id,
+                        };
+
+                        Box::pin(db_utils::try_redis_get_else_try_database_get(
+                            async {
+                                kv_wrapper(
+                                    self,
+                                    KvOperation::<storage_types::PaymentMethod>::HGet(
+                                        &lookup.sk_id,
+                                    ),
+                                    key,
+                                )
+                                .await?
+                                .try_into_hget()
+                            },
+                            database_call,
+                        ))
+                        .await
+                    }
                 }
-            }
+            };
+
+            get_pm()
+                .await?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
+
         // not supported in kv
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn get_payment_method_count_by_customer_id_merchant_id_status(
             &self,
@@ -207,17 +375,25 @@ mod storage {
         #[instrument(skip_all)]
         async fn insert_payment_method(
             &self,
-            mut payment_method_new: storage_types::PaymentMethodNew,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
             storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
             let storage_scheme = decide_storage_scheme::<_, storage_types::PaymentMethod>(
                 self,
                 storage_scheme,
                 Op::Insert,
             )
             .await;
+
+            let mut payment_method_new = payment_method
+                .construct_new()
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+
             payment_method_new.update_storage_scheme(storage_scheme);
-            match storage_scheme {
+            let pm = match storage_scheme {
                 MerchantStorageScheme::PostgresOnly => {
                     let conn = connection::pg_connection_write(self).await?;
                     payment_method_new
@@ -231,11 +407,10 @@ mod storage {
 
                     let key = PartitionKey::MerchantIdCustomerId {
                         merchant_id: &merchant_id,
-                        customer_id: customer_id.get_string_repr(),
+                        customer_id: &customer_id,
                     };
                     let key_str = key.to_string();
-                    let field =
-                        format!("payment_method_id_{}", payment_method_new.payment_method_id);
+                    let field = format!("payment_method_id_{}", payment_method_new.get_id());
 
                     let reverse_lookup_entry = |v: String| diesel_models::ReverseLookupNew {
                         sk_id: field.clone(),
@@ -245,8 +420,7 @@ mod storage {
                         updated_by: storage_scheme.to_string(),
                     };
 
-                    let lookup_id1 =
-                        format!("payment_method_{}", &payment_method_new.payment_method_id);
+                    let lookup_id1 = format!("payment_method_{}", payment_method_new.get_id());
                     let mut reverse_lookups = vec![lookup_id1];
                     if let Some(locker_id) = &payment_method_new.locker_id {
                         reverse_lookups.push(format!("payment_method_locker_{}", locker_id))
@@ -281,37 +455,55 @@ mod storage {
                     {
                         Ok(HsetnxReply::KeyNotSet) => Err(errors::StorageError::DuplicateValue {
                             entity: "payment_method",
-                            key: Some(storage_payment_method.payment_method_id),
+                            key: Some(storage_payment_method.get_id().clone()),
                         }
                         .into()),
                         Ok(HsetnxReply::KeySet) => Ok(storage_payment_method),
                         Err(er) => Err(er).change_context(errors::StorageError::KVError),
                     }
                 }
-            }
+            }?;
+
+            pm.convert(
+                state,
+                key_store.key.get_inner(),
+                key_store.merchant_id.clone().into(),
+            )
+            .await
+            .change_context(errors::StorageError::DecryptionError)
         }
 
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn update_payment_method(
             &self,
-            payment_method: storage_types::PaymentMethod,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
             payment_method_update: storage_types::PaymentMethodUpdate,
             storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let payment_method = Conversion::convert(payment_method)
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+
             let merchant_id = payment_method.merchant_id.clone();
             let customer_id = payment_method.customer_id.clone();
             let key = PartitionKey::MerchantIdCustomerId {
                 merchant_id: &merchant_id,
-                customer_id: customer_id.get_string_repr(),
+                customer_id: &customer_id,
             };
-            let field = format!("payment_method_id_{}", payment_method.payment_method_id);
+            let field = format!("payment_method_id_{}", payment_method.get_id());
             let storage_scheme = decide_storage_scheme::<_, storage_types::PaymentMethod>(
                 self,
                 storage_scheme,
                 Op::Update(key.clone(), &field, payment_method.updated_by.as_deref()),
             )
             .await;
-            match storage_scheme {
+            let pm = match storage_scheme {
                 MerchantStorageScheme::PostgresOnly => {
                     let conn = connection::pg_connection_write(self).await?;
                     payment_method
@@ -359,36 +551,155 @@ mod storage {
 
                     Ok(updated_payment_method)
                 }
-            }
+            }?;
+
+            pm.convert(
+                state,
+                key_store.key.get_inner(),
+                key_store.merchant_id.clone().into(),
+            )
+            .await
+            .change_context(errors::StorageError::DecryptionError)
         }
 
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        #[instrument(skip_all)]
+        async fn update_payment_method(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
+            payment_method_update: storage_types::PaymentMethodUpdate,
+            storage_scheme: MerchantStorageScheme,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let payment_method = Conversion::convert(payment_method)
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+
+            let merchant_id = payment_method.merchant_id.clone();
+            let customer_id = payment_method.customer_id.clone();
+            let key = PartitionKey::MerchantIdCustomerId {
+                merchant_id: &merchant_id,
+                customer_id: &customer_id,
+            };
+            let field = format!("payment_method_id_{}", payment_method.get_id());
+            let storage_scheme = decide_storage_scheme::<_, storage_types::PaymentMethod>(
+                self,
+                storage_scheme,
+                Op::Update(key.clone(), &field, payment_method.updated_by.as_deref()),
+            )
+            .await;
+            let pm = match storage_scheme {
+                MerchantStorageScheme::PostgresOnly => {
+                    let conn = connection::pg_connection_write(self).await?;
+                    payment_method
+                        .update_with_id(
+                            &conn,
+                            payment_method_update.convert_to_payment_method_update(storage_scheme),
+                        )
+                        .await
+                        .map_err(|error| report!(errors::StorageError::from(error)))
+                }
+                MerchantStorageScheme::RedisKv => {
+                    let key_str = key.to_string();
+
+                    let p_update: PaymentMethodUpdateInternal =
+                        payment_method_update.convert_to_payment_method_update(storage_scheme);
+                    let updated_payment_method =
+                        p_update.clone().apply_changeset(payment_method.clone());
+
+                    let redis_value = serde_json::to_string(&updated_payment_method)
+                        .change_context(errors::StorageError::SerializationFailed)?;
+
+                    let redis_entry = kv::TypedSql {
+                        op: kv::DBOperation::Update {
+                            updatable: kv::Updateable::PaymentMethodUpdate(
+                                kv::PaymentMethodUpdateMems {
+                                    orig: payment_method,
+                                    update_data: p_update,
+                                },
+                            ),
+                        },
+                    };
+
+                    kv_wrapper::<(), _, _>(
+                        self,
+                        KvOperation::<diesel_models::PaymentMethod>::Hset(
+                            (&field, redis_value),
+                            redis_entry,
+                        ),
+                        key,
+                    )
+                    .await
+                    .map_err(|err| err.to_redis_failed_response(&key_str))?
+                    .try_into_hset()
+                    .change_context(errors::StorageError::KVError)?;
+
+                    Ok(updated_payment_method)
+                }
+            }?;
+
+            pm.convert(
+                state,
+                key_store.key.get_inner(),
+                key_store.merchant_id.clone().into(),
+            )
+            .await
+            .change_context(errors::StorageError::DecryptionError)
+        }
+
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn find_payment_method_by_customer_id_merchant_id_list(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             customer_id: &id_type::CustomerId,
             merchant_id: &id_type::MerchantId,
             limit: Option<i64>,
-        ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError> {
+        ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
-            storage_types::PaymentMethod::find_by_customer_id_merchant_id(
+            let payment_methods = storage_types::PaymentMethod::find_by_customer_id_merchant_id(
                 &conn,
                 customer_id,
                 merchant_id,
                 limit,
             )
             .await
-            .map_err(|error| report!(errors::StorageError::from(error)))
+            .map_err(|error| report!(errors::StorageError::from(error)))?;
+
+            let pm_futures = payment_methods
+                .into_iter()
+                .map(|pm| async {
+                    pm.convert(
+                        state,
+                        key_store.key.get_inner(),
+                        key_store.merchant_id.clone().into(),
+                    )
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)
+                })
+                .collect::<Vec<_>>();
+
+            let domain_payment_methods = futures::future::try_join_all(pm_futures).await?;
+
+            Ok(domain_payment_methods)
         }
 
         #[instrument(skip_all)]
         async fn find_payment_method_by_customer_id_merchant_id_status(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             customer_id: &id_type::CustomerId,
             merchant_id: &id_type::MerchantId,
             status: common_enums::PaymentMethodStatus,
             limit: Option<i64>,
             storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError> {
+        ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
             let database_call = || async {
                 storage_types::PaymentMethod::find_by_customer_id_merchant_id_status(
@@ -402,12 +713,12 @@ mod storage {
                 .map_err(|error| report!(errors::StorageError::from(error)))
             };
 
-            match storage_scheme {
+            let payment_methods = match storage_scheme {
                 MerchantStorageScheme::PostgresOnly => database_call().await,
                 MerchantStorageScheme::RedisKv => {
                     let key = PartitionKey::MerchantIdCustomerId {
                         merchant_id,
-                        customer_id: customer_id.get_string_repr(),
+                        customer_id,
                     };
 
                     let pattern = "payment_method_id_*";
@@ -435,14 +746,37 @@ mod storage {
                     ))
                     .await
                 }
-            }
+            }?;
+
+            let pm_futures = payment_methods
+                .into_iter()
+                .map(|pm| async {
+                    pm.convert(
+                        state,
+                        key_store.key.get_inner(),
+                        key_store.merchant_id.clone().into(),
+                    )
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)
+                })
+                .collect::<Vec<_>>();
+
+            let domain_payment_methods = futures::future::try_join_all(pm_futures).await?;
+
+            Ok(domain_payment_methods)
         }
 
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         async fn delete_payment_method_by_merchant_id_payment_method_id(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             merchant_id: &id_type::MerchantId,
             payment_method_id: &str,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
             let conn = connection::pg_connection_write(self).await?;
             storage_types::PaymentMethod::delete_by_merchant_id_payment_method_id(
                 &conn,
@@ -450,15 +784,72 @@ mod storage {
                 payment_method_id,
             )
             .await
-            .map_err(|error| report!(errors::StorageError::from(error)))
+            .map_err(|error| report!(errors::StorageError::from(error)))?
+            .convert(
+                state,
+                key_store.key.get_inner(),
+                key_store.merchant_id.clone().into(),
+            )
+            .await
+            .change_context(errors::StorageError::DecryptionError)
+        }
+
+        // Soft delete, Check if KV stuff is needed here
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        async fn delete_payment_method(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let payment_method = Conversion::convert(payment_method)
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+            let conn = connection::pg_connection_write(self).await?;
+            let payment_method_update = storage_types::PaymentMethodUpdate::StatusUpdate {
+                status: Some(common_enums::PaymentMethodStatus::Inactive),
+            };
+            payment_method
+                .update_with_id(&conn, payment_method_update.into())
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }
+
+        // Check if KV stuff is needed here
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        async fn find_payment_method_by_fingerprint_id(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            fingerprint_id: &str,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            storage_types::PaymentMethod::find_by_fingerprint_id(&conn, fingerprint_id)
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
     }
 }
 
 #[cfg(not(feature = "kv_store"))]
 mod storage {
-    use common_utils::id_type;
-    use error_stack::report;
+    use common_utils::{id_type, types::keymanager::KeyManagerState};
+    use error_stack::{report, ResultExt};
+    use hyperswitch_domain_models::behaviour::{Conversion, ReverseConversion};
     use router_env::{instrument, tracing};
 
     use super::PaymentMethodInterface;
@@ -466,35 +857,89 @@ mod storage {
         connection,
         core::errors::{self, CustomResult},
         services::Store,
-        types::storage::{self as storage_types, enums::MerchantStorageScheme},
+        types::{
+            domain,
+            storage::{self as storage_types, enums::MerchantStorageScheme},
+        },
     };
 
     #[async_trait::async_trait]
     impl PaymentMethodInterface for Store {
         #[instrument(skip_all)]
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         async fn find_payment_method(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             payment_method_id: &str,
             _storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
             storage_types::PaymentMethod::find_by_payment_method_id(&conn, payment_method_id)
                 .await
-                .map_err(|error| report!(errors::StorageError::from(error)))
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
 
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        async fn find_payment_method(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method_id: &str,
+            _storage_scheme: MerchantStorageScheme,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            storage_types::PaymentMethod::find_by_id(&conn, payment_method_id)
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }
+
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn find_payment_method_by_locker_id(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             locker_id: &str,
             _storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
             storage_types::PaymentMethod::find_by_locker_id(&conn, locker_id)
                 .await
-                .map_err(|error| report!(errors::StorageError::from(error)))
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
 
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn get_payment_method_count_by_customer_id_merchant_id_status(
             &self,
@@ -516,74 +961,187 @@ mod storage {
         #[instrument(skip_all)]
         async fn insert_payment_method(
             &self,
-            payment_method_new: storage_types::PaymentMethodNew,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
             _storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let payment_method_new = payment_method
+                .construct_new()
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+
             let conn = connection::pg_connection_write(self).await?;
             payment_method_new
                 .insert(&conn)
                 .await
-                .map_err(|error| report!(errors::StorageError::from(error)))
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
 
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn update_payment_method(
             &self,
-            payment_method: storage_types::PaymentMethod,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
             payment_method_update: storage_types::PaymentMethodUpdate,
             _storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let payment_method = Conversion::convert(payment_method)
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+
             let conn = connection::pg_connection_write(self).await?;
             payment_method
                 .update_with_payment_method_id(&conn, payment_method_update.into())
                 .await
-                .map_err(|error| report!(errors::StorageError::from(error)))
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }
+
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        #[instrument(skip_all)]
+        async fn update_payment_method(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
+            payment_method_update: storage_types::PaymentMethodUpdate,
+            _storage_scheme: MerchantStorageScheme,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let payment_method = payment_method
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+
+            let conn = connection::pg_connection_write(self).await?;
+            payment_method
+                .update_with_id(&conn, payment_method_update.into())
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
 
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         #[instrument(skip_all)]
         async fn find_payment_method_by_customer_id_merchant_id_list(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             customer_id: &id_type::CustomerId,
             merchant_id: &id_type::MerchantId,
             limit: Option<i64>,
-        ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError> {
+        ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
-            storage_types::PaymentMethod::find_by_customer_id_merchant_id(
+            let payment_methods = storage_types::PaymentMethod::find_by_customer_id_merchant_id(
                 &conn,
                 customer_id,
                 merchant_id,
                 limit,
             )
             .await
-            .map_err(|error| report!(errors::StorageError::from(error)))
+            .map_err(|error| report!(errors::StorageError::from(error)))?;
+
+            let pm_futures = payment_methods
+                .into_iter()
+                .map(|pm| async {
+                    pm.convert(
+                        state,
+                        key_store.key.get_inner(),
+                        key_store.merchant_id.clone().into(),
+                    )
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)
+                })
+                .collect::<Vec<_>>();
+
+            let domain_payment_methods = futures::future::try_join_all(pm_futures).await?;
+
+            Ok(domain_payment_methods)
         }
 
         #[instrument(skip_all)]
         async fn find_payment_method_by_customer_id_merchant_id_status(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             customer_id: &id_type::CustomerId,
             merchant_id: &id_type::MerchantId,
             status: common_enums::PaymentMethodStatus,
             limit: Option<i64>,
             _storage_scheme: MerchantStorageScheme,
-        ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError> {
+        ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
             let conn = connection::pg_connection_read(self).await?;
-            storage_types::PaymentMethod::find_by_customer_id_merchant_id_status(
-                &conn,
-                customer_id,
-                merchant_id,
-                status,
-                limit,
-            )
-            .await
-            .map_err(|error| report!(errors::StorageError::from(error)))
+            let payment_methods =
+                storage_types::PaymentMethod::find_by_customer_id_merchant_id_status(
+                    &conn,
+                    customer_id,
+                    merchant_id,
+                    status,
+                    limit,
+                )
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?;
+
+            let pm_futures = payment_methods
+                .into_iter()
+                .map(|pm| async {
+                    pm.convert(
+                        state,
+                        key_store.key.get_inner(),
+                        key_store.merchant_id.clone().into(),
+                    )
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)
+                })
+                .collect::<Vec<_>>();
+
+            let domain_payment_methods = futures::future::try_join_all(pm_futures).await?;
+
+            Ok(domain_payment_methods)
         }
 
+        #[cfg(all(
+            any(feature = "v1", feature = "v2"),
+            not(feature = "payment_methods_v2")
+        ))]
         async fn delete_payment_method_by_merchant_id_payment_method_id(
             &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
             merchant_id: &id_type::MerchantId,
             payment_method_id: &str,
-        ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
             let conn = connection::pg_connection_write(self).await?;
             storage_types::PaymentMethod::delete_by_merchant_id_payment_method_id(
                 &conn,
@@ -591,7 +1149,61 @@ mod storage {
                 payment_method_id,
             )
             .await
-            .map_err(|error| report!(errors::StorageError::from(error)))
+            .map_err(|error| report!(errors::StorageError::from(error)))?
+            .convert(
+                state,
+                key_store.key.get_inner(),
+                key_store.merchant_id.clone().into(),
+            )
+            .await
+            .change_context(errors::StorageError::DecryptionError)
+        }
+
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        async fn delete_payment_method(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            payment_method: domain::PaymentMethod,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let payment_method = Conversion::convert(payment_method)
+                .await
+                .change_context(errors::StorageError::DecryptionError)?;
+            let conn = connection::pg_connection_write(self).await?;
+            let payment_method_update = storage_types::PaymentMethodUpdate::StatusUpdate {
+                status: Some(common_enums::PaymentMethodStatus::Inactive),
+            };
+            payment_method
+                .update_with_id(&conn, payment_method_update.into())
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }
+
+        #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+        async fn find_payment_method_by_fingerprint_id(
+            &self,
+            state: &KeyManagerState,
+            key_store: &domain::MerchantKeyStore,
+            fingerprint_id: &str,
+        ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+            let conn = connection::pg_connection_read(self).await?;
+            storage_types::PaymentMethod::find_by_fingerprint_id(&conn, fingerprint_id)
+                .await
+                .map_err(|error| report!(errors::StorageError::from(error)))?
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)
         }
     }
 }
@@ -600,17 +1212,26 @@ mod storage {
 impl PaymentMethodInterface for MockDb {
     async fn find_payment_method(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         payment_method_id: &str,
         _storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         let payment_methods = self.payment_methods.lock().await;
         let payment_method = payment_methods
             .iter()
-            .find(|pm| pm.payment_method_id == payment_method_id)
+            .find(|pm| pm.get_id() == payment_method_id)
             .cloned();
 
         match payment_method {
-            Some(pm) => Ok(pm),
+            Some(pm) => Ok(pm
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)?),
             None => Err(errors::StorageError::ValueNotFound(
                 "cannot find payment method".to_string(),
             )
@@ -618,11 +1239,17 @@ impl PaymentMethodInterface for MockDb {
         }
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn find_payment_method_by_locker_id(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         locker_id: &str,
         _storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         let payment_methods = self.payment_methods.lock().await;
         let payment_method = payment_methods
             .iter()
@@ -630,7 +1257,14 @@ impl PaymentMethodInterface for MockDb {
             .cloned();
 
         match payment_method {
-            Some(pm) => Ok(pm),
+            Some(pm) => Ok(pm
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)?),
             None => Err(errors::StorageError::ValueNotFound(
                 "cannot find payment method".to_string(),
             )
@@ -638,6 +1272,10 @@ impl PaymentMethodInterface for MockDb {
         }
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn get_payment_method_count_by_customer_id_merchant_id_status(
         &self,
         customer_id: &id_type::CustomerId,
@@ -658,53 +1296,33 @@ impl PaymentMethodInterface for MockDb {
 
     async fn insert_payment_method(
         &self,
-        payment_method_new: storage_types::PaymentMethodNew,
+        _state: &KeyManagerState,
+        _key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
         _storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         let mut payment_methods = self.payment_methods.lock().await;
 
-        let payment_method = storage_types::PaymentMethod {
-            customer_id: payment_method_new.customer_id,
-            merchant_id: payment_method_new.merchant_id,
-            payment_method_id: payment_method_new.payment_method_id,
-            locker_id: payment_method_new.locker_id,
-            accepted_currency: payment_method_new.accepted_currency,
-            scheme: payment_method_new.scheme,
-            token: payment_method_new.token,
-            cardholder_name: payment_method_new.cardholder_name,
-            issuer_name: payment_method_new.issuer_name,
-            issuer_country: payment_method_new.issuer_country,
-            payer_country: payment_method_new.payer_country,
-            is_stored: payment_method_new.is_stored,
-            swift_code: payment_method_new.swift_code,
-            direct_debit_token: payment_method_new.direct_debit_token,
-            created_at: payment_method_new.created_at,
-            last_modified: payment_method_new.last_modified,
-            payment_method: payment_method_new.payment_method,
-            payment_method_type: payment_method_new.payment_method_type,
-            payment_method_issuer: payment_method_new.payment_method_issuer,
-            payment_method_issuer_code: payment_method_new.payment_method_issuer_code,
-            metadata: payment_method_new.metadata,
-            payment_method_data: payment_method_new.payment_method_data,
-            last_used_at: payment_method_new.last_used_at,
-            connector_mandate_details: payment_method_new.connector_mandate_details,
-            customer_acceptance: payment_method_new.customer_acceptance,
-            status: payment_method_new.status,
-            client_secret: payment_method_new.client_secret,
-            network_transaction_id: payment_method_new.network_transaction_id,
-            updated_by: payment_method_new.updated_by,
-            payment_method_billing_address: payment_method_new.payment_method_billing_address,
-        };
-        payment_methods.push(payment_method.clone());
+        let pm = Conversion::convert(payment_method.clone())
+            .await
+            .change_context(errors::StorageError::DecryptionError)?;
+
+        payment_methods.push(pm);
         Ok(payment_method)
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn find_payment_method_by_customer_id_merchant_id_list(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         customer_id: &id_type::CustomerId,
         merchant_id: &id_type::MerchantId,
         _limit: Option<i64>,
-    ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError> {
+    ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
         let payment_methods = self.payment_methods.lock().await;
         let payment_methods_found: Vec<storage_types::PaymentMethod> = payment_methods
             .iter()
@@ -718,18 +1336,35 @@ impl PaymentMethodInterface for MockDb {
                     .into(),
             )
         } else {
-            Ok(payment_methods_found)
+            let pm_futures = payment_methods_found
+                .into_iter()
+                .map(|pm| async {
+                    pm.convert(
+                        state,
+                        key_store.key.get_inner(),
+                        key_store.merchant_id.clone().into(),
+                    )
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)
+                })
+                .collect::<Vec<_>>();
+
+            let domain_payment_methods = futures::future::try_join_all(pm_futures).await?;
+
+            Ok(domain_payment_methods)
         }
     }
 
     async fn find_payment_method_by_customer_id_merchant_id_status(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         customer_id: &id_type::CustomerId,
         merchant_id: &id_type::MerchantId,
         status: common_enums::PaymentMethodStatus,
         _limit: Option<i64>,
         _storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<Vec<storage_types::PaymentMethod>, errors::StorageError> {
+    ) -> CustomResult<Vec<domain::PaymentMethod>, errors::StorageError> {
         let payment_methods = self.payment_methods.lock().await;
         let payment_methods_found: Vec<storage_types::PaymentMethod> = payment_methods
             .iter()
@@ -747,22 +1382,51 @@ impl PaymentMethodInterface for MockDb {
                     .into(),
             )
         } else {
-            Ok(payment_methods_found)
+            let pm_futures = payment_methods_found
+                .into_iter()
+                .map(|pm| async {
+                    pm.convert(
+                        state,
+                        key_store.key.get_inner(),
+                        key_store.merchant_id.clone().into(),
+                    )
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)
+                })
+                .collect::<Vec<_>>();
+
+            let domain_payment_methods = futures::future::try_join_all(pm_futures).await?;
+
+            Ok(domain_payment_methods)
         }
     }
 
+    #[cfg(all(
+        any(feature = "v1", feature = "v2"),
+        not(feature = "payment_methods_v2")
+    ))]
     async fn delete_payment_method_by_merchant_id_payment_method_id(
         &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
         merchant_id: &id_type::MerchantId,
         payment_method_id: &str,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         let mut payment_methods = self.payment_methods.lock().await;
-        match payment_methods.iter().position(|pm| {
-            pm.merchant_id == *merchant_id && pm.payment_method_id == payment_method_id
-        }) {
+        match payment_methods
+            .iter()
+            .position(|pm| pm.merchant_id == *merchant_id && pm.get_id() == payment_method_id)
+        {
             Some(index) => {
                 let deleted_payment_method = payment_methods.remove(index);
-                Ok(deleted_payment_method)
+                Ok(deleted_payment_method
+                    .convert(
+                        state,
+                        key_store.key.get_inner(),
+                        key_store.merchant_id.clone().into(),
+                    )
+                    .await
+                    .change_context(errors::StorageError::DecryptionError)?)
             }
             None => Err(errors::StorageError::ValueNotFound(
                 "cannot find payment method to delete".to_string(),
@@ -773,16 +1437,18 @@ impl PaymentMethodInterface for MockDb {
 
     async fn update_payment_method(
         &self,
-        payment_method: storage_types::PaymentMethod,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
         payment_method_update: storage_types::PaymentMethodUpdate,
         _storage_scheme: MerchantStorageScheme,
-    ) -> CustomResult<storage_types::PaymentMethod, errors::StorageError> {
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
         let pm_update_res = self
             .payment_methods
             .lock()
             .await
             .iter_mut()
-            .find(|pm| pm.payment_method_id == payment_method.payment_method_id)
+            .find(|pm| pm.get_id() == payment_method.get_id())
             .map(|pm| {
                 let payment_method_updated =
                     PaymentMethodUpdateInternal::from(payment_method_update)
@@ -792,11 +1458,88 @@ impl PaymentMethodInterface for MockDb {
             });
 
         match pm_update_res {
-            Some(result) => Ok(result),
+            Some(result) => Ok(result
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)?),
             None => Err(errors::StorageError::ValueNotFound(
                 "cannot find payment method to update".to_string(),
             )
             .into()),
         }
     }
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    async fn delete_payment_method(
+        &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        payment_method: domain::PaymentMethod,
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+        let payment_method_update = storage_types::PaymentMethodUpdate::StatusUpdate {
+            status: Some(common_enums::PaymentMethodStatus::Inactive),
+        };
+
+        let pm_update_res = self
+            .payment_methods
+            .lock()
+            .await
+            .iter_mut()
+            .find(|pm| pm.get_id() == payment_method.get_id())
+            .map(|pm| {
+                let payment_method_updated =
+                    PaymentMethodUpdateInternal::from(payment_method_update)
+                        .create_payment_method(pm.clone());
+                *pm = payment_method_updated.clone();
+                payment_method_updated
+            });
+
+        match pm_update_res {
+            Some(result) => Ok(result
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)?),
+            None => Err(errors::StorageError::ValueNotFound(
+                "cannot find payment method to update".to_string(),
+            )
+            .into()),
+        }
+    }
+
+    #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+    async fn find_payment_method_by_fingerprint_id(
+        &self,
+        state: &KeyManagerState,
+        key_store: &domain::MerchantKeyStore,
+        fingerprint_id: &str,
+    ) -> CustomResult<domain::PaymentMethod, errors::StorageError> {
+        let payment_methods = self.payment_methods.lock().await;
+        let payment_method = payment_methods
+            .iter()
+            .find(|pm| pm.locker_fingerprint_id == Some(fingerprint_id.to_string()))
+            .cloned();
+
+        match payment_method {
+            Some(pm) => Ok(pm
+                .convert(
+                    state,
+                    key_store.key.get_inner(),
+                    key_store.merchant_id.clone().into(),
+                )
+                .await
+                .change_context(errors::StorageError::DecryptionError)?),
+            None => Err(errors::StorageError::ValueNotFound(
+                "cannot find payment method".to_string(),
+            )
+            .into()),
+        }
+    }
 }
diff --git a/crates/router/src/types/api/mandates.rs b/crates/router/src/types/api/mandates.rs
index bf8f00526e9..9262877a385 100644
--- a/crates/router/src/types/api/mandates.rs
+++ b/crates/router/src/types/api/mandates.rs
@@ -32,6 +32,91 @@ pub(crate) trait MandateResponseExt: Sized {
     ) -> RouterResult<Self>;
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
+#[async_trait::async_trait]
+impl MandateResponseExt for MandateResponse {
+    async fn from_db_mandate(
+        state: &SessionState,
+        key_store: domain::MerchantKeyStore,
+        mandate: storage::Mandate,
+        storage_scheme: storage_enums::MerchantStorageScheme,
+    ) -> RouterResult<Self> {
+        let db = &*state.store;
+        let payment_method = db
+            .find_payment_method(
+                &(state.into()),
+                &key_store,
+                &mandate.payment_method_id,
+                storage_scheme,
+            )
+            .await
+            .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
+
+        let pm = payment_method
+            .payment_method
+            .get_required_value("payment_method")
+            .change_context(errors::ApiErrorResponse::PaymentMethodNotFound)
+            .attach_printable("payment_method not found")?;
+
+        let card = if pm == storage_enums::PaymentMethod::Card {
+            // if locker is disabled , decrypt the payment method data
+            let card_details = if state.conf.locker.locker_enabled {
+                let card = payment_methods::cards::get_card_from_locker(
+                    state,
+                    &payment_method.customer_id,
+                    &payment_method.merchant_id,
+                    payment_method
+                        .locker_id
+                        .as_ref()
+                        .unwrap_or(payment_method.get_id()),
+                )
+                .await?;
+
+                payment_methods::transformers::get_card_detail(&payment_method, card)
+                    .change_context(errors::ApiErrorResponse::InternalServerError)
+                    .attach_printable("Failed while getting card details")?
+            } else {
+                payment_methods::cards::get_card_details_without_locker_fallback(
+                    &payment_method,
+                    state,
+                )
+                .await?
+            };
+
+            Some(MandateCardDetails::from(card_details).into_inner())
+        } else {
+            None
+        };
+        let payment_method_type = payment_method
+            .payment_method_type
+            .map(|pmt| pmt.to_string());
+        Ok(Self {
+            mandate_id: mandate.mandate_id,
+            customer_acceptance: Some(api::payments::CustomerAcceptance {
+                acceptance_type: if mandate.customer_ip_address.is_some() {
+                    api::payments::AcceptanceType::Online
+                } else {
+                    api::payments::AcceptanceType::Offline
+                },
+                accepted_at: mandate.customer_accepted_at,
+                online: Some(api::payments::OnlineMandate {
+                    ip_address: mandate.customer_ip_address,
+                    user_agent: mandate.customer_user_agent.unwrap_or_default(),
+                }),
+            }),
+            card,
+            status: mandate.mandate_status,
+            payment_method: pm.to_string(),
+            payment_method_type,
+            payment_method_id: mandate.payment_method_id,
+        })
+    }
+}
+
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
 #[async_trait::async_trait]
 impl MandateResponseExt for MandateResponse {
     async fn from_db_mandate(
@@ -42,7 +127,12 @@ impl MandateResponseExt for MandateResponse {
     ) -> RouterResult<Self> {
         let db = &*state.store;
         let payment_method = db
-            .find_payment_method(&mandate.payment_method_id, storage_scheme)
+            .find_payment_method(
+                &(state.into()),
+                &key_store,
+                &mandate.payment_method_id,
+                storage_scheme,
+            )
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentMethodNotFound)?;
 
@@ -62,7 +152,7 @@ impl MandateResponseExt for MandateResponse {
                     payment_method
                         .locker_id
                         .as_ref()
-                        .unwrap_or(&payment_method.payment_method_id),
+                        .unwrap_or(&payment_method.id),
                 )
                 .await?;
 
@@ -73,7 +163,6 @@ impl MandateResponseExt for MandateResponse {
                 payment_methods::cards::get_card_details_without_locker_fallback(
                     &payment_method,
                     state,
-                    &key_store,
                 )
                 .await?
             };
diff --git a/crates/router/src/types/domain.rs b/crates/router/src/types/domain.rs
index d1dad53ce34..37b8d2b8016 100644
--- a/crates/router/src/types/domain.rs
+++ b/crates/router/src/types/domain.rs
@@ -25,6 +25,12 @@ mod merchant_connector_account;
 mod merchant_key_store {
     pub use hyperswitch_domain_models::merchant_key_store::MerchantKeyStore;
 }
+pub mod payment_methods {
+    pub use hyperswitch_domain_models::payment_methods::*;
+}
+pub mod consts {
+    pub use hyperswitch_domain_models::consts::*;
+}
 pub mod payments;
 pub mod types;
 #[cfg(feature = "olap")]
@@ -33,9 +39,11 @@ pub mod user_key_store;
 
 pub use address::*;
 pub use business_profile::*;
+pub use consts::*;
 pub use event::*;
 pub use merchant_connector_account::*;
 pub use merchant_key_store::*;
+pub use payment_methods::*;
 pub use payments::*;
 #[cfg(feature = "olap")]
 pub use user::*;
diff --git a/crates/router/src/types/transformers.rs b/crates/router/src/types/transformers.rs
index ca4ba575d79..7b637b76c16 100644
--- a/crates/router/src/types/transformers.rs
+++ b/crates/router/src/types/transformers.rs
@@ -88,19 +88,19 @@ impl ForeignFrom<api_models::refunds::RefundType> for storage_enums::RefundType
 impl
     ForeignFrom<(
         Option<payment_methods::CardDetailFromLocker>,
-        diesel_models::PaymentMethod,
+        domain::PaymentMethod,
     )> for payment_methods::PaymentMethodResponse
 {
     fn foreign_from(
         (card_details, item): (
             Option<payment_methods::CardDetailFromLocker>,
-            diesel_models::PaymentMethod,
+            domain::PaymentMethod,
         ),
     ) -> Self {
         Self {
-            merchant_id: item.merchant_id,
-            customer_id: Some(item.customer_id),
-            payment_method_id: item.payment_method_id,
+            merchant_id: item.merchant_id.to_owned(),
+            customer_id: Some(item.customer_id.to_owned()),
+            payment_method_id: item.get_id().clone(),
             payment_method: item.payment_method,
             payment_method_type: item.payment_method_type,
             card: card_details,
@@ -121,23 +121,22 @@ impl
 impl
     ForeignFrom<(
         Option<payment_methods::CardDetailFromLocker>,
-        diesel_models::PaymentMethod,
+        domain::PaymentMethod,
     )> for payment_methods::PaymentMethodResponse
 {
     fn foreign_from(
         (card_details, item): (
             Option<payment_methods::CardDetailFromLocker>,
-            diesel_models::PaymentMethod,
+            domain::PaymentMethod,
         ),
     ) -> Self {
         Self {
-            merchant_id: item.merchant_id,
-            customer_id: item.customer_id,
-            payment_method_id: item.payment_method_id,
+            merchant_id: item.merchant_id.to_owned(),
+            customer_id: item.customer_id.to_owned(),
+            payment_method_id: item.get_id().clone(),
             payment_method: item.payment_method,
             payment_method_type: item.payment_method_type,
-            payment_method_data: card_details
-                .map(|card| payment_methods::PaymentMethodResponseData::Card(card.clone())),
+            payment_method_data: card_details.map(payment_methods::PaymentMethodResponseData::Card),
             recurring_enabled: false,
             metadata: item.metadata,
             created: Some(item.created_at),
diff --git a/crates/router/src/workflows/payment_method_status_update.rs b/crates/router/src/workflows/payment_method_status_update.rs
index 5120f1ba197..abc30074cda 100644
--- a/crates/router/src/workflows/payment_method_status_update.rs
+++ b/crates/router/src/workflows/payment_method_status_update.rs
@@ -45,7 +45,12 @@ impl ProcessTrackerWorkflow<SessionState> for PaymentMethodStatusUpdateWorkflow
             .await?;
 
         let payment_method = db
-            .find_payment_method(&pm_id, merchant_account.storage_scheme)
+            .find_payment_method(
+                &(state.into()),
+                &key_store,
+                &pm_id,
+                merchant_account.storage_scheme,
+            )
             .await?;
 
         if payment_method.status != prev_pm_status {
@@ -61,7 +66,13 @@ impl ProcessTrackerWorkflow<SessionState> for PaymentMethodStatusUpdateWorkflow
         };
 
         let res = db
-            .update_payment_method(payment_method, pm_update, merchant_account.storage_scheme)
+            .update_payment_method(
+                &(state.into()),
+                &key_store,
+                payment_method,
+                pm_update,
+                merchant_account.storage_scheme,
+            )
             .await
             .map_err(errors::ProcessTrackerError::EStorageError);
 
diff --git a/crates/storage_impl/Cargo.toml b/crates/storage_impl/Cargo.toml
index 557f6969fb7..22badd47cd0 100644
--- a/crates/storage_impl/Cargo.toml
+++ b/crates/storage_impl/Cargo.toml
@@ -16,6 +16,7 @@ v1 = ["api_models/v1", "diesel_models/v1", "hyperswitch_domain_models/v1"]
 v2 = ["api_models/v2", "diesel_models/v2", "hyperswitch_domain_models/v2"]
 payment_v2 = ["hyperswitch_domain_models/payment_v2", "diesel_models/payment_v2"]
 customer_v2 = ["api_models/customer_v2", "diesel_models/customer_v2", "hyperswitch_domain_models/customer_v2"]
+payment_methods_v2 = ["diesel_models/payment_methods_v2", "api_models/payment_methods_v2", "hyperswitch_domain_models/payment_methods_v2"]
 
 [dependencies]
 # First Party dependencies
diff --git a/crates/storage_impl/src/lib.rs b/crates/storage_impl/src/lib.rs
index 995f11b9cc6..c00d2b445f8 100644
--- a/crates/storage_impl/src/lib.rs
+++ b/crates/storage_impl/src/lib.rs
@@ -406,6 +406,10 @@ impl UniqueConstraints for diesel_models::PayoutAttempt {
     }
 }
 
+#[cfg(all(
+    any(feature = "v1", feature = "v2"),
+    not(feature = "payment_methods_v2")
+))]
 impl UniqueConstraints for diesel_models::PaymentMethod {
     fn unique_constraints(&self) -> Vec<String> {
         vec![format!("paymentmethod_{}", self.payment_method_id)]
@@ -415,6 +419,16 @@ impl UniqueConstraints for diesel_models::PaymentMethod {
     }
 }
 
+#[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
+impl UniqueConstraints for diesel_models::PaymentMethod {
+    fn unique_constraints(&self) -> Vec<String> {
+        vec![format!("paymentmethod_{}", self.id)]
+    }
+    fn table_name(&self) -> &str {
+        "PaymentMethod"
+    }
+}
+
 impl UniqueConstraints for diesel_models::Mandate {
     fn unique_constraints(&self) -> Vec<String> {
         vec![format!(
diff --git a/crates/storage_impl/src/redis/kv_store.rs b/crates/storage_impl/src/redis/kv_store.rs
index 8b5674084a2..8511702baa2 100644
--- a/crates/storage_impl/src/redis/kv_store.rs
+++ b/crates/storage_impl/src/redis/kv_store.rs
@@ -32,7 +32,7 @@ pub enum PartitionKey<'a> {
     },
     MerchantIdCustomerId {
         merchant_id: &'a common_utils::id_type::MerchantId,
-        customer_id: &'a str,
+        customer_id: &'a common_utils::id_type::CustomerId,
     },
     #[cfg(all(feature = "v2", feature = "customer_v2"))]
     MerchantIdMerchantReferenceId {
@@ -73,8 +73,9 @@ impl<'a> std::fmt::Display for PartitionKey<'a> {
                 merchant_id,
                 customer_id,
             } => f.write_str(&format!(
-                "mid_{}_cust_{customer_id}",
-                merchant_id.get_string_repr()
+                "mid_{}_cust_{}",
+                merchant_id.get_string_repr(),
+                customer_id.get_string_repr()
             )),
             #[cfg(all(feature = "v2", feature = "customer_v2"))]
             PartitionKey::MerchantIdMerchantReferenceId {
diff --git a/migrations/2024-09-02-112941_add_version_in_payment_methods/down.sql b/migrations/2024-09-02-112941_add_version_in_payment_methods/down.sql
new file mode 100644
index 00000000000..e3635ea68f0
--- /dev/null
+++ b/migrations/2024-09-02-112941_add_version_in_payment_methods/down.sql
@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS version;
\ No newline at end of file
diff --git a/migrations/2024-09-02-112941_add_version_in_payment_methods/up.sql b/migrations/2024-09-02-112941_add_version_in_payment_methods/up.sql
new file mode 100644
index 00000000000..70d8c9d34b7
--- /dev/null
+++ b/migrations/2024-09-02-112941_add_version_in_payment_methods/up.sql
@@ -0,0 +1,3 @@
+-- Your SQL goes here
+ALTER TABLE payment_methods
+ADD COLUMN IF NOT EXISTS version "ApiVersion" NOT NULL DEFAULT 'v1';
\ No newline at end of file
diff --git a/v2_migrations/2024-08-23-112510_payment_methods_v2_db_changes/down.sql b/v2_migrations/2024-08-23-112510_payment_methods_v2_db_changes/down.sql
new file mode 100644
index 00000000000..cde77d754f8
--- /dev/null
+++ b/v2_migrations/2024-08-23-112510_payment_methods_v2_db_changes/down.sql
@@ -0,0 +1,44 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS accepted_currency "Currency"[];
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS scheme VARCHAR(32);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS token VARCHAR(128);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS cardholder_name VARCHAR(255);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS issuer_name VARCHAR(64);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS issuer_country VARCHAR(64);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS is_stored BOOLEAN;
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS direct_debit_token VARCHAR(128);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS swift_code VARCHAR(32);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS payment_method_issuer VARCHAR(128);
+
+CREATE TYPE "PaymentMethodIssuerCode" AS ENUM (
+    'jp_hdfc',
+    'jp_icici',
+    'jp_googlepay',
+    'jp_applepay',
+    'jp_phonepe',
+    'jp_wechat',
+    'jp_sofort',
+    'jp_giropay',
+    'jp_sepa',
+    'jp_bacs'
+);
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS payment_method_issuer_code "PaymentMethodIssuerCode";
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS locker_fingerprint_id;
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS payment_method_id VARCHAR(64);
+UPDATE payment_methods SET payment_method_id = id;
+ALTER TABLE payment_methods DROP CONSTRAINT IF EXISTS payment_methods_pkey;
+ALTER TABLE payment_methods ADD CONSTRAINT payment_methods_pkey PRIMARY KEY (payment_method_id);
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS id;
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS id SERIAL;
\ No newline at end of file
diff --git a/v2_migrations/2024-08-23-112510_payment_methods_v2_db_changes/up.sql b/v2_migrations/2024-08-23-112510_payment_methods_v2_db_changes/up.sql
new file mode 100644
index 00000000000..151cf068cff
--- /dev/null
+++ b/v2_migrations/2024-08-23-112510_payment_methods_v2_db_changes/up.sql
@@ -0,0 +1,35 @@
+-- Your SQL goes here
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS accepted_currency;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS scheme;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS token;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS cardholder_name;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS issuer_name;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS issuer_country;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS payer_country;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS is_stored;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS direct_debit_token;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS swift_code;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS payment_method_issuer;
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS payment_method_issuer_code;
+
+DROP TYPE IF EXISTS "PaymentMethodIssuerCode";
+
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS locker_fingerprint_id VARCHAR(64);
+
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS id;
+ALTER TABLE payment_methods ADD COLUMN IF NOT EXISTS id VARCHAR(64);
+UPDATE payment_methods SET id = payment_method_id;
+ALTER TABLE payment_methods DROP CONSTRAINT IF EXISTS payment_methods_pkey;
+ALTER TABLE payment_methods ADD CONSTRAINT payment_methods_pkey PRIMARY KEY (id);
+ALTER TABLE payment_methods DROP COLUMN IF EXISTS payment_method_id;
\ 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 -->
- Diesel model for payment methods v2
- Domain models for payment methods v1/v2

### Additional Changes

- [ ] This PR modifies the API contract
- [x] This PR modifies the database schema
- [ ] 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)?
-->
- Test all flows (payments/payment_methods) for v1
- v2 cannot be tested as core logic is incomplete

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