Bug: feat(vsaas): add processor_merchant_id and created_by column in payment_intents and payments_attempts for v1



Tracking issue #1: https://github.com/juspay/hyperswitch-control-center/issues/2544
Tracking issue #2: https://github.com/juspay/hyperswitch-control-center/issues/2546
diff --git a/crates/common_utils/src/macros.rs b/crates/common_utils/src/macros.rs
index 2ab0560cd41..8dcffbb6643 100644
--- a/crates/common_utils/src/macros.rs
+++ b/crates/common_utils/src/macros.rs
@@ -437,3 +437,387 @@ macro_rules! type_name {
             .unwrap_or_default();
     };
 }
+
+/// **Note** Creates an enum wrapper that implements `FromStr`, `Display`, `Serialize`, and `Deserialize`
+/// based on a specific string representation format: `"VariantName<delimiter>FieldValue"`.
+/// It handles parsing errors by returning a dedicated `Invalid` variant.
+/// *Note*: The macro adds `Invalid,` automatically.
+///
+/// # Use Case
+///
+/// This macro is designed for scenarios where you need an enum, with each variant
+/// holding a single piece of associated data, to be easily convertible to and from
+/// a simple string format. This is useful for cases where enum is serialized to key value pairs
+///
+/// It avoids more complex serialization structures (like JSON objects `{"VariantName": value}`)
+/// in favor of a plain string representation.
+///
+/// # Input Enum Format and Constraints
+///
+/// To use this macro, the enum definition must adhere to the following structure:
+///
+/// 1.  **Public Enum:** The enum must be declared as `pub enum EnumName { ... }`.
+/// 2.  **Struct Variants Only:** All variants must be struct variants (using `{}`).
+/// 3.  **Exactly One Field:** Each struct variant must contain *exactly one* named field.
+///     * **Valid:** `VariantA { value: i32 }`
+///     * **Invalid:** `VariantA(i32)` (tuple variant)
+///     * **Invalid:** `VariantA` or `VariantA {}` (no field)
+///     * **Invalid:** `VariantA { value: i32, other: bool }` (multiple fields)
+/// 4.  **Tag Delimiter:** The macro invocation must specify a `tag_delimeter` literal,
+///     which is the character used to separate the variant name from the field data in
+///     the string representation (e.g., `tag_delimeter = ":",`).
+/// 5.  **Field Type Requirements:** The type of the single field in each variant (`$field_ty`)
+///     must implement:
+///     * `core::str::FromStr`: To parse the field's data from the string part.
+///       The `Err` type should ideally be convertible to a meaningful error, though the
+///       macro currently uses a generic error message upon failure.
+///     * `core::fmt::Display`: To convert the field's data into the string part.
+///     * `serde::Serialize` and `serde::Deserialize<'de>`: Although the macro implements
+///       custom `Serialize`/`Deserialize` for the *enum* using the string format, the field
+///       type itself must satisfy these bounds if required elsewhere or by generic contexts.
+///       The macro's implementations rely solely on `Display` and `FromStr` for the conversion.
+/// 6.  **Error Type:** This macro uses `core::convert::Infallible` as it never fails but gives
+///     `Self::Invalid` variant.
+///
+/// # Serialization and Deserialization (`serde`)
+///
+/// When `serde` features are enabled and the necessary traits are derived or implemented,
+/// this macro implements `Serialize` and `Deserialize` for the enum:
+///
+/// **Serialization:** An enum value like `MyEnum::VariantA { value: 123 }` (with `tag_delimeter = ":",`)
+///     will be serialized into the string `"VariantA:123"`. If serializing to JSON, this results
+///     in a JSON string: `"\"VariantA:123\""`.
+/// **Deserialization:** The macro expects a string matching the format `"VariantName<delimiter>FieldValue"`.
+///     It uses the enum's `FromStr` implementation internally. When deserializing from JSON, it
+///     expects a JSON string containing the correctly formatted value (e.g., `"\"VariantA:123\""`).
+///
+/// # `Display` and `FromStr`
+///
+/// **`Display`:** Formats valid variants to `"VariantName<delimiter>FieldValue"` and catch-all cases to `"Invalid"`.
+/// **`FromStr`:** Parses `"VariantName<delimiter>FieldValue"` to the variant, or returns `Self::Invalid`
+///   if the input string is malformed or `"Invalid"`.
+///
+/// # Example
+///
+/// ```rust
+/// use std::str::FromStr;
+///
+/// crate::impl_enum_str!(
+///     tag_delimeter = ":",
+///     #[derive(Debug, PartialEq, Clone)] // Add other derives as needed
+///     pub enum Setting {
+///         Timeout { duration_ms: u32 },
+///         Username { name: String },
+///     }
+/// );
+/// // Note: The macro adds `Invalid,` automatically.
+///
+/// fn main() {
+///     // Display
+///     let setting1 = Setting::Timeout { duration_ms: 5000 };
+///     assert_eq!(setting1.to_string(), "Timeout:5000");
+///     assert_eq!(Setting::Invalid.to_string(), "Invalid");
+///
+///     // FromStr (returns Self, not Result)
+///     let parsed_setting: Setting = "Username:admin".parse().expect("Valid parse"); // parse() itself doesn't panic
+///     assert_eq!(parsed_setting, Setting::Username { name: "admin".to_string() });
+///
+///     let invalid_format: Setting = "Timeout".parse().expect("Parse always returns Self");
+///     assert_eq!(invalid_format, Setting::Invalid); // Malformed input yields Invalid
+///
+///     let bad_data: Setting = "Timeout:fast".parse().expect("Parse always returns Self");
+///     assert_eq!(bad_data, Setting::Invalid); // Bad field data yields Invalid
+///
+///     let unknown_tag: Setting = "Unknown:abc".parse().expect("Parse always returns Self");
+///     assert_eq!(unknown_tag, Setting::Invalid); // Unknown tag yields Invalid
+///
+///     let explicit_invalid: Setting = "Invalid".parse().expect("Parse always returns Self");
+///     assert_eq!(explicit_invalid, Setting::Invalid); // "Invalid" string yields Invalid
+///
+///     // Serde (requires derive Serialize/Deserialize on Setting)
+///     // let json_output = serde_json::to_string(&setting1).unwrap();
+///     // assert_eq!(json_output, "\"Timeout:5000\"");
+///     // let invalid_json_output = serde_json::to_string(&Setting::Invalid).unwrap();
+///     // assert_eq!(invalid_json_output, "\"Invalid\"");
+///
+///     // let deserialized: Setting = serde_json::from_str("\"Username:guest\"").unwrap();
+///     // assert_eq!(deserialized, Setting::Username { name: "guest".to_string() });
+///     // let deserialized_invalid: Setting = serde_json::from_str("\"Invalid\"").unwrap();
+///     // assert_eq!(deserialized_invalid, Setting::Invalid);
+///     // let deserialized_malformed: Setting = serde_json::from_str("\"TimeoutFast\"").unwrap();
+///     // assert_eq!(deserialized_malformed, Setting::Invalid); // Malformed -> Invalid
+/// }
+///
+/// # // Mock macro definition for doctest purposes
+/// # #[macro_export] macro_rules! impl_enum_str { ($($tt:tt)*) => { $($tt)* } }
+/// ```
+#[macro_export]
+macro_rules! impl_enum_str {
+    (
+        tag_delimeter = $tag_delim:literal,
+        $(#[$enum_attr:meta])*
+        pub enum $enum_name:ident {
+            $(
+                $(#[$variant_attr:meta])*
+                $variant:ident {
+                    $(#[$field_attr:meta])*
+                    $field:ident : $field_ty:ty $(,)?
+                }
+            ),* $(,)?
+        }
+    ) => {
+        $(#[$enum_attr])*
+        pub enum $enum_name {
+            $(
+                $(#[$variant_attr])*
+                $variant {
+                    $(#[$field_attr])*
+                    $field : $field_ty
+                },
+            )*
+            /// Represents a parsing failure.
+            Invalid, // Automatically add the Invalid variant
+        }
+
+        // Implement FromStr - now returns Self, not Result
+        impl core::str::FromStr for $enum_name {
+            // No associated error type needed
+            type Err = core::convert::Infallible; // FromStr requires an Err type, use Infallible
+
+            fn from_str(s: &str) -> Result<Self, Self::Err> {
+                // Check for explicit "Invalid" string first
+                if s == "Invalid" {
+                    return Ok(Self::Invalid);
+                }
+
+                let Some((tag, associated_data)) = s.split_once($tag_delim) else {
+                    // Missing delimiter -> Invalid
+                    return Ok(Self::Invalid);
+                };
+
+                let result = match tag {
+                    $(
+                        stringify!($variant) => {
+                            // Try to parse the field data
+                            match associated_data.parse::<$field_ty>() {
+                                Ok(parsed_field) => {
+                                    // Success -> construct the variant
+                                     Self::$variant { $field: parsed_field }
+                                },
+                                Err(_) => {
+                                     // Field parse failure -> Invalid
+                                     Self::Invalid
+                                }
+                            }
+                        }
+                    ),*
+                    // Unknown tag -> Invalid
+                    _ => Self::Invalid,
+                };
+                Ok(result) // Always Ok because failure modes return Self::Invalid
+            }
+        }
+
+        // Implement Serialize
+        impl ::serde::Serialize for $enum_name {
+            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+            where
+                S: ::serde::Serializer,
+            {
+                match self {
+                    $(
+                        Self::$variant { $field } => {
+                            let s = format!("{}{}{}", stringify!($variant), $tag_delim, $field);
+                            serializer.serialize_str(&s)
+                        }
+                    )*
+                    // Handle Invalid variant
+                    Self::Invalid => serializer.serialize_str("Invalid"),
+                }
+            }
+        }
+
+        // Implement Deserialize
+        impl<'de> ::serde::Deserialize<'de> for $enum_name {
+            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+            where
+                D: ::serde::Deserializer<'de>,
+            {
+                struct EnumVisitor;
+
+                impl<'de> ::serde::de::Visitor<'de> for EnumVisitor {
+                    type Value = $enum_name;
+
+                    fn expecting(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+                        formatter.write_str(concat!("a string like VariantName", $tag_delim, "field_data or 'Invalid'"))
+                    }
+
+                    // Leverage the FromStr implementation which now returns Self::Invalid on failure
+                    fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
+                    where
+                        E: ::serde::de::Error,
+                    {
+                        // parse() now returns Result<Self, Infallible>
+                        // We unwrap() the Ok because it's infallible.
+                       Ok(value.parse::<$enum_name>().unwrap())
+                    }
+
+                     fn visit_string<E>(self, value: String) -> Result<Self::Value, E>
+                    where
+                        E: ::serde::de::Error,
+                    {
+                        Ok(value.parse::<$enum_name>().unwrap())
+                    }
+                }
+
+                deserializer.deserialize_str(EnumVisitor)
+            }
+        }
+
+        // Implement Display
+        impl core::fmt::Display for $enum_name {
+            fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+                match self {
+                    $(
+                        Self::$variant { $field } => {
+                            write!(f, "{}{}{}", stringify!($variant), $tag_delim, $field)
+                        }
+                    )*
+                     // Handle Invalid variant
+                    Self::Invalid => write!(f, "Invalid"),
+                }
+            }
+        }
+    };
+}
+
+// --- Tests ---
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::panic, clippy::expect_used)]
+
+    use serde_json::{json, Value as JsonValue};
+
+    use crate::impl_enum_str;
+
+    impl_enum_str!(
+        tag_delimeter = ":",
+        #[derive(Debug, PartialEq, Clone)]
+        pub enum TestEnum {
+            VariantA { value: i32 },
+            VariantB { text: String },
+            VariantC { id: u64 },
+            VariantJson { data: JsonValue },
+        } // Note: Invalid variant is added automatically by the macro
+    );
+
+    #[test]
+    fn test_enum_from_str_ok() {
+        // Success cases just parse directly
+        let parsed_a: TestEnum = "VariantA:42".parse().unwrap(); // Unwrapping Infallible is fine
+        assert_eq!(parsed_a, TestEnum::VariantA { value: 42 });
+
+        let parsed_b: TestEnum = "VariantB:hello world".parse().unwrap();
+        assert_eq!(
+            parsed_b,
+            TestEnum::VariantB {
+                text: "hello world".to_string()
+            }
+        );
+
+        let parsed_c: TestEnum = "VariantC:123456789012345".parse().unwrap();
+        assert_eq!(
+            parsed_c,
+            TestEnum::VariantC {
+                id: 123456789012345
+            }
+        );
+
+        let parsed_json: TestEnum = r#"VariantJson:{"ok":true}"#.parse().unwrap();
+        assert_eq!(
+            parsed_json,
+            TestEnum::VariantJson {
+                data: json!({"ok": true})
+            }
+        );
+    }
+
+    #[test]
+    fn test_enum_from_str_failures_yield_invalid() {
+        // Missing delimiter
+        let parsed: TestEnum = "VariantA".parse().unwrap();
+        assert_eq!(parsed, TestEnum::Invalid);
+
+        // Unknown tag
+        let parsed: TestEnum = "UnknownVariant:123".parse().unwrap();
+        assert_eq!(parsed, TestEnum::Invalid);
+
+        // Bad field data for i32
+        let parsed: TestEnum = "VariantA:not_a_number".parse().unwrap();
+        assert_eq!(parsed, TestEnum::Invalid);
+
+        // Bad field data for JsonValue
+        let parsed: TestEnum = r#"VariantJson:{"bad_json"#.parse().unwrap();
+        assert_eq!(parsed, TestEnum::Invalid);
+
+        // Empty field data for non-string (e.g., i32)
+        let parsed: TestEnum = "VariantA:".parse().unwrap();
+        assert_eq!(parsed, TestEnum::Invalid);
+
+        // Empty field data for string IS valid for String type
+        let parsed_str: TestEnum = "VariantB:".parse().unwrap();
+        assert_eq!(
+            parsed_str,
+            TestEnum::VariantB {
+                text: "".to_string()
+            }
+        );
+
+        // Parsing the literal "Invalid" string
+        let parsed_invalid_str: TestEnum = "Invalid".parse().unwrap();
+        assert_eq!(parsed_invalid_str, TestEnum::Invalid);
+    }
+
+    #[test]
+    fn test_enum_display_and_serialize() {
+        // Display valid
+        let value_a = TestEnum::VariantA { value: 99 };
+        assert_eq!(value_a.to_string(), "VariantA:99");
+        // Serialize valid
+        let json_a = serde_json::to_string(&value_a).expect("Serialize A failed");
+        assert_eq!(json_a, "\"VariantA:99\""); // Serializes to JSON string
+
+        // Display Invalid
+        let value_invalid = TestEnum::Invalid;
+        assert_eq!(value_invalid.to_string(), "Invalid");
+        // Serialize Invalid
+        let json_invalid = serde_json::to_string(&value_invalid).expect("Serialize Invalid failed");
+        assert_eq!(json_invalid, "\"Invalid\""); // Serializes to JSON string "Invalid"
+    }
+
+    #[test]
+    fn test_enum_deserialize() {
+        // Deserialize valid
+        let input_a = "\"VariantA:123\"";
+        let deserialized_a: TestEnum = serde_json::from_str(input_a).expect("Deserialize A failed");
+        assert_eq!(deserialized_a, TestEnum::VariantA { value: 123 });
+
+        // Deserialize explicit "Invalid"
+        let input_invalid = "\"Invalid\"";
+        let deserialized_invalid: TestEnum =
+            serde_json::from_str(input_invalid).expect("Deserialize Invalid failed");
+        assert_eq!(deserialized_invalid, TestEnum::Invalid);
+
+        // Deserialize malformed string (according to macro rules) -> Invalid
+        let input_malformed = "\"VariantA_no_delimiter\"";
+        let deserialized_malformed: TestEnum =
+            serde_json::from_str(input_malformed).expect("Deserialize malformed should succeed");
+        assert_eq!(deserialized_malformed, TestEnum::Invalid);
+
+        // Deserialize string with bad field data -> Invalid
+        let input_bad_data = "\"VariantA:not_a_number\"";
+        let deserialized_bad_data: TestEnum =
+            serde_json::from_str(input_bad_data).expect("Deserialize bad data should succeed");
+        assert_eq!(deserialized_bad_data, TestEnum::Invalid);
+    }
+}
diff --git a/crates/common_utils/src/types.rs b/crates/common_utils/src/types.rs
index 4b24cf281cb..e2cc195fe05 100644
--- a/crates/common_utils/src/types.rs
+++ b/crates/common_utils/src/types.rs
@@ -49,6 +49,7 @@ use crate::{
     },
     errors::{CustomResult, ParsingError, PercentageError, ValidationError},
     fp_utils::when,
+    impl_enum_str,
 };
 
 /// Represents Percentage Value between 0 and 100 both inclusive
@@ -1348,3 +1349,22 @@ where
         self.0.to_sql(out)
     }
 }
+
+impl_enum_str!(
+    tag_delimeter = ":",
+    /// CreatedBy conveys the information about the creator (identifier) as well as the origin or
+    /// trigger (Api, Jwt) of the record.
+    #[derive(Eq, PartialEq, Debug, Clone)]
+    pub enum CreatedBy {
+        /// Api variant
+        Api {
+            /// merchant id of creator.
+            merchant_id: String,
+        },
+        /// Jwt variant
+        Jwt {
+            /// user id of creator.
+            user_id: String,
+        },
+    }
+);
diff --git a/crates/diesel_models/src/payment_attempt.rs b/crates/diesel_models/src/payment_attempt.rs
index 63f039c32f9..0a3c44deee1 100644
--- a/crates/diesel_models/src/payment_attempt.rs
+++ b/crates/diesel_models/src/payment_attempt.rs
@@ -89,6 +89,8 @@ pub struct PaymentAttempt {
     pub capture_before: Option<PrimitiveDateTime>,
     pub card_discovery: Option<storage_enums::CardDiscovery>,
     pub charges: Option<common_types::payments::ConnectorChargeResponseData>,
+    pub processor_merchant_id: Option<id_type::MerchantId>,
+    pub created_by: Option<String>,
     pub payment_method_type_v2: storage_enums::PaymentMethod,
     pub connector_payment_id: Option<ConnectorTransactionId>,
     pub payment_method_subtype: storage_enums::PaymentMethodType,
@@ -197,6 +199,8 @@ pub struct PaymentAttempt {
     pub charges: Option<common_types::payments::ConnectorChargeResponseData>,
     pub issuer_error_code: Option<String>,
     pub issuer_error_message: Option<String>,
+    pub processor_merchant_id: Option<id_type::MerchantId>,
+    pub created_by: Option<String>,
     pub setup_future_usage_applied: Option<storage_enums::FutureUsage>,
 }
 
@@ -334,6 +338,8 @@ pub struct PaymentAttemptNew {
     pub network_decline_code: Option<String>,
     pub network_advice_code: Option<String>,
     pub network_error_message: Option<String>,
+    pub processor_merchant_id: Option<id_type::MerchantId>,
+    pub created_by: Option<String>,
 }
 
 #[cfg(feature = "v1")]
@@ -411,6 +417,8 @@ pub struct PaymentAttemptNew {
     #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
     pub capture_before: Option<PrimitiveDateTime>,
     pub card_discovery: Option<storage_enums::CardDiscovery>,
+    pub processor_merchant_id: Option<id_type::MerchantId>,
+    pub created_by: Option<String>,
     pub setup_future_usage_applied: Option<storage_enums::FutureUsage>,
 }
 
diff --git a/crates/diesel_models/src/payment_intent.rs b/crates/diesel_models/src/payment_intent.rs
index 7476639a47d..840c8e085b9 100644
--- a/crates/diesel_models/src/payment_intent.rs
+++ b/crates/diesel_models/src/payment_intent.rs
@@ -60,6 +60,8 @@ pub struct PaymentIntent {
     pub platform_merchant_id: Option<common_utils::id_type::MerchantId>,
     pub force_3ds_challenge: Option<bool>,
     pub force_3ds_challenge_trigger: Option<bool>,
+    pub processor_merchant_id: Option<common_utils::id_type::MerchantId>,
+    pub created_by: Option<String>,
     pub merchant_reference_id: Option<common_utils::id_type::PaymentReferenceId>,
     pub billing_address: Option<Encryption>,
     pub shipping_address: Option<Encryption>,
@@ -148,6 +150,8 @@ pub struct PaymentIntent {
     pub platform_merchant_id: Option<common_utils::id_type::MerchantId>,
     pub force_3ds_challenge: Option<bool>,
     pub force_3ds_challenge_trigger: Option<bool>,
+    pub processor_merchant_id: Option<common_utils::id_type::MerchantId>,
+    pub created_by: Option<String>,
 }
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize, diesel::AsExpression, PartialEq)]
@@ -337,6 +341,8 @@ pub struct PaymentIntentNew {
     pub platform_merchant_id: Option<common_utils::id_type::MerchantId>,
     pub force_3ds_challenge: Option<bool>,
     pub force_3ds_challenge_trigger: Option<bool>,
+    pub processor_merchant_id: Option<common_utils::id_type::MerchantId>,
+    pub created_by: Option<String>,
 }
 
 #[cfg(feature = "v1")]
@@ -408,6 +414,8 @@ pub struct PaymentIntentNew {
     pub platform_merchant_id: Option<common_utils::id_type::MerchantId>,
     pub force_3ds_challenge: Option<bool>,
     pub force_3ds_challenge_trigger: Option<bool>,
+    pub processor_merchant_id: Option<common_utils::id_type::MerchantId>,
+    pub created_by: Option<String>,
 }
 
 #[cfg(feature = "v2")]
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 802aa698a82..374b341a47e 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -933,6 +933,10 @@ diesel::table! {
         #[max_length = 64]
         issuer_error_code -> Nullable<Varchar>,
         issuer_error_message -> Nullable<Text>,
+        #[max_length = 64]
+        processor_merchant_id -> Nullable<Varchar>,
+        #[max_length = 255]
+        created_by -> Nullable<Varchar>,
         setup_future_usage_applied -> Nullable<FutureUsage>,
     }
 }
@@ -1021,6 +1025,10 @@ diesel::table! {
         platform_merchant_id -> Nullable<Varchar>,
         force_3ds_challenge -> Nullable<Bool>,
         force_3ds_challenge_trigger -> Nullable<Bool>,
+        #[max_length = 64]
+        processor_merchant_id -> Nullable<Varchar>,
+        #[max_length = 255]
+        created_by -> Nullable<Varchar>,
     }
 }
 
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index b9ca1c5da46..9a72f663e4f 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -874,6 +874,10 @@ diesel::table! {
         capture_before -> Nullable<Timestamp>,
         card_discovery -> Nullable<CardDiscovery>,
         charges -> Nullable<Jsonb>,
+        #[max_length = 64]
+        processor_merchant_id -> Nullable<Varchar>,
+        #[max_length = 255]
+        created_by -> Nullable<Varchar>,
         payment_method_type_v2 -> Varchar,
         #[max_length = 128]
         connector_payment_id -> Nullable<Varchar>,
@@ -954,6 +958,10 @@ diesel::table! {
         force_3ds_challenge -> Nullable<Bool>,
         force_3ds_challenge_trigger -> Nullable<Bool>,
         #[max_length = 64]
+        processor_merchant_id -> Nullable<Varchar>,
+        #[max_length = 255]
+        created_by -> Nullable<Varchar>,
+        #[max_length = 64]
         merchant_reference_id -> Nullable<Varchar>,
         billing_address -> Nullable<Bytea>,
         shipping_address -> Nullable<Bytea>,
diff --git a/crates/diesel_models/src/user/sample_data.rs b/crates/diesel_models/src/user/sample_data.rs
index 9212d416cb7..3779877abcb 100644
--- a/crates/diesel_models/src/user/sample_data.rs
+++ b/crates/diesel_models/src/user/sample_data.rs
@@ -210,6 +210,8 @@ pub struct PaymentAttemptBatchNew {
     pub extended_authorization_applied: Option<ExtendedAuthorizationAppliedBool>,
     pub capture_before: Option<PrimitiveDateTime>,
     pub card_discovery: Option<common_enums::CardDiscovery>,
+    pub processor_merchant_id: Option<common_utils::id_type::MerchantId>,
+    pub created_by: Option<String>,
     pub setup_future_usage_applied: Option<common_enums::FutureUsage>,
 }
 
@@ -293,6 +295,8 @@ impl PaymentAttemptBatchNew {
             extended_authorization_applied: self.extended_authorization_applied,
             capture_before: self.capture_before,
             card_discovery: self.card_discovery,
+            processor_merchant_id: self.processor_merchant_id,
+            created_by: self.created_by,
             setup_future_usage_applied: self.setup_future_usage_applied,
         }
     }
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index 90ba708c109..35b9f4a4613 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -15,7 +15,7 @@ use common_utils::{
     errors::CustomResult,
     ext_traits::ValueExt,
     id_type, pii,
-    types::{keymanager::ToEncryptable, MinorUnit},
+    types::{keymanager::ToEncryptable, CreatedBy, MinorUnit},
 };
 use diesel_models::payment_intent::TaxDetails;
 #[cfg(feature = "v2")]
@@ -115,7 +115,8 @@ pub struct PaymentIntent {
     pub skip_external_tax_calculation: Option<bool>,
     pub request_extended_authorization: Option<RequestExtendedAuthorizationBool>,
     pub psd2_sca_exemption_type: Option<storage_enums::ScaExemptionType>,
-    pub platform_merchant_id: Option<id_type::MerchantId>,
+    pub processor_merchant_id: id_type::MerchantId,
+    pub created_by: Option<CreatedBy>,
     pub force_3ds_challenge: Option<bool>,
     pub force_3ds_challenge_trigger: Option<bool>,
 }
@@ -493,13 +494,15 @@ pub struct PaymentIntent {
     pub payment_link_config: Option<diesel_models::PaymentLinkConfigRequestForPayments>,
     /// The straight through routing algorithm id that is used for this payment. This overrides the default routing algorithm that is configured in business profile.
     pub routing_algorithm_id: Option<id_type::RoutingId>,
-    /// Identifier for the platform merchant.
-    pub platform_merchant_id: Option<id_type::MerchantId>,
     /// Split Payment Data
     pub split_payments: Option<common_types::payments::SplitPaymentsRequest>,
 
     pub force_3ds_challenge: Option<bool>,
     pub force_3ds_challenge_trigger: Option<bool>,
+    /// merchant who owns the credentials of the processor, i.e. processor owner
+    pub processor_merchant_id: id_type::MerchantId,
+    /// merchantwho invoked the resource based api (identifier) and through what source (Api, Jwt(Dashboard))
+    pub created_by: Option<CreatedBy>,
 }
 
 #[cfg(feature = "v2")]
@@ -655,11 +658,11 @@ impl PaymentIntent {
                 .payment_link_config
                 .map(ApiModelToDieselModelConvertor::convert_from),
             routing_algorithm_id: request.routing_algorithm_id,
-            platform_merchant_id: platform_merchant_id
-                .map(|merchant_account| merchant_account.get_id().to_owned()),
             split_payments: None,
             force_3ds_challenge: None,
             force_3ds_challenge_trigger: None,
+            processor_merchant_id: merchant_account.get_id().clone(),
+            created_by: None,
         })
     }
 
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
index a074cca9b5d..77183dcf0d7 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
@@ -15,7 +15,7 @@ use common_utils::{
     id_type, pii,
     types::{
         keymanager::{self, KeyManagerState},
-        ConnectorTransactionId, ConnectorTransactionIdTrait, MinorUnit,
+        ConnectorTransactionId, ConnectorTransactionIdTrait, CreatedBy, MinorUnit,
     },
 };
 use diesel_models::{
@@ -454,6 +454,10 @@ pub struct PaymentAttempt {
     pub charges: Option<common_types::payments::ConnectorChargeResponseData>,
     /// Additional data that might be required by hyperswitch, to enable some specific features.
     pub feature_metadata: Option<PaymentAttemptFeatureMetadata>,
+    /// merchant who owns the credentials of the processor, i.e. processor owner
+    pub processor_merchant_id: id_type::MerchantId,
+    /// merchantwho invoked the resource based api (identifier) and through what source (Api, Jwt(Dashboard))
+    pub created_by: Option<CreatedBy>,
 }
 
 impl PaymentAttempt {
@@ -583,6 +587,8 @@ impl PaymentAttempt {
             id,
             card_discovery: None,
             feature_metadata: None,
+            processor_merchant_id: payment_intent.merchant_id.clone(),
+            created_by: None,
         })
     }
 
@@ -670,6 +676,8 @@ impl PaymentAttempt {
             feature_metadata: None,
             id,
             card_discovery: None,
+            processor_merchant_id: payment_intent.merchant_id.clone(),
+            created_by: None,
         })
     }
 
@@ -767,6 +775,8 @@ impl PaymentAttempt {
             }),
             card_discovery: None,
             charges: None,
+            processor_merchant_id: payment_intent.merchant_id.clone(),
+            created_by: None,
         })
     }
 
@@ -859,6 +869,10 @@ pub struct PaymentAttempt {
     pub charges: Option<common_types::payments::ConnectorChargeResponseData>,
     pub issuer_error_code: Option<String>,
     pub issuer_error_message: Option<String>,
+    /// merchant who owns the credentials of the processor, i.e. processor owner
+    pub processor_merchant_id: id_type::MerchantId,
+    /// merchantwho invoked the resource based api (identifier) and through what source (Api, Jwt(Dashboard))
+    pub created_by: Option<CreatedBy>,
     pub setup_future_usage_applied: Option<storage_enums::FutureUsage>,
 }
 
@@ -1109,6 +1123,10 @@ pub struct PaymentAttemptNew {
     pub extended_authorization_applied: Option<ExtendedAuthorizationAppliedBool>,
     pub capture_before: Option<PrimitiveDateTime>,
     pub card_discovery: Option<common_enums::CardDiscovery>,
+    /// merchant who owns the credentials of the processor, i.e. processor owner
+    pub processor_merchant_id: id_type::MerchantId,
+    /// merchantwho invoked the resource based api (identifier) and through what source (Api, Jwt(Dashboard))
+    pub created_by: Option<CreatedBy>,
     pub setup_future_usage_applied: Option<storage_enums::FutureUsage>,
 }
 
@@ -1859,6 +1877,8 @@ impl behaviour::Conversion for PaymentAttempt {
             setup_future_usage_applied: self.setup_future_usage_applied,
             // Below fields are deprecated. Please add any new fields above this line.
             connector_transaction_data: None,
+            processor_merchant_id: Some(self.processor_merchant_id),
+            created_by: self.created_by.map(|cb| cb.to_string()),
         })
     }
 
@@ -1877,7 +1897,7 @@ impl behaviour::Conversion for PaymentAttempt {
                 .cloned();
             Ok::<Self, error_stack::Report<common_utils::errors::CryptoError>>(Self {
                 payment_id: storage_model.payment_id,
-                merchant_id: storage_model.merchant_id,
+                merchant_id: storage_model.merchant_id.clone(),
                 attempt_id: storage_model.attempt_id,
                 status: storage_model.status,
                 net_amount: NetAmount::new(
@@ -1947,6 +1967,12 @@ impl behaviour::Conversion for PaymentAttempt {
                 charges: storage_model.charges,
                 issuer_error_code: storage_model.issuer_error_code,
                 issuer_error_message: storage_model.issuer_error_message,
+                processor_merchant_id: storage_model
+                    .processor_merchant_id
+                    .unwrap_or(storage_model.merchant_id),
+                created_by: storage_model
+                    .created_by
+                    .and_then(|created_by| created_by.parse::<CreatedBy>().ok()),
                 setup_future_usage_applied: storage_model.setup_future_usage_applied,
             })
         }
@@ -2033,6 +2059,8 @@ impl behaviour::Conversion for PaymentAttempt {
             extended_authorization_applied: self.extended_authorization_applied,
             capture_before: self.capture_before,
             card_discovery: self.card_discovery,
+            processor_merchant_id: Some(self.processor_merchant_id),
+            created_by: self.created_by.map(|cb| cb.to_string()),
             setup_future_usage_applied: self.setup_future_usage_applied,
         })
     }
@@ -2103,6 +2131,8 @@ impl behaviour::Conversion for PaymentAttempt {
             card_discovery,
             charges,
             feature_metadata,
+            processor_merchant_id,
+            created_by,
         } = self;
 
         let AttemptAmountDetails {
@@ -2195,6 +2225,8 @@ impl behaviour::Conversion for PaymentAttempt {
             network_error_message: error
                 .as_ref()
                 .and_then(|details| details.network_error_message.clone()),
+            processor_merchant_id: Some(processor_merchant_id),
+            created_by: created_by.map(|cb| cb.to_string()),
         })
     }
 
@@ -2266,7 +2298,7 @@ impl behaviour::Conversion for PaymentAttempt {
 
             Ok::<Self, error_stack::Report<common_utils::errors::CryptoError>>(Self {
                 payment_id: storage_model.payment_id,
-                merchant_id: storage_model.merchant_id,
+                merchant_id: storage_model.merchant_id.clone(),
                 id: storage_model.id,
                 status: storage_model.status,
                 amount_details,
@@ -2311,6 +2343,12 @@ impl behaviour::Conversion for PaymentAttempt {
                 connector_token_details: storage_model.connector_token_details,
                 card_discovery: storage_model.card_discovery,
                 feature_metadata: storage_model.feature_metadata.map(From::from),
+                processor_merchant_id: storage_model
+                    .processor_merchant_id
+                    .unwrap_or(storage_model.merchant_id),
+                created_by: storage_model
+                    .created_by
+                    .and_then(|created_by| created_by.parse::<CreatedBy>().ok()),
             })
         }
         .await
@@ -2367,6 +2405,8 @@ impl behaviour::Conversion for PaymentAttempt {
             card_discovery,
             charges,
             feature_metadata,
+            processor_merchant_id,
+            created_by,
         } = self;
 
         let card_network = payment_method_data
@@ -2456,6 +2496,8 @@ impl behaviour::Conversion for PaymentAttempt {
             network_error_message: error_details
                 .as_ref()
                 .and_then(|details| details.network_error_message.clone()),
+            processor_merchant_id: Some(processor_merchant_id),
+            created_by: created_by.map(|cb| cb.to_string()),
         })
     }
 }
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
index a55b1b62fab..de2e0dd46c9 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
@@ -14,7 +14,7 @@ use common_utils::{
     type_name,
     types::{
         keymanager::{self, KeyManagerState, ToEncryptable},
-        MinorUnit,
+        CreatedBy, MinorUnit,
     },
 };
 #[cfg(feature = "v2")]
@@ -1602,10 +1602,11 @@ impl behaviour::Conversion for PaymentIntent {
             customer_present,
             routing_algorithm_id,
             payment_link_config,
-            platform_merchant_id,
             split_payments,
             force_3ds_challenge,
             force_3ds_challenge_trigger,
+            processor_merchant_id,
+            created_by,
         } = self;
         Ok(DieselPaymentIntent {
             skip_external_tax_calculation: Some(amount_details.get_external_tax_action_as_bool()),
@@ -1684,10 +1685,12 @@ impl behaviour::Conversion for PaymentIntent {
             routing_algorithm_id,
             psd2_sca_exemption_type: None,
             request_extended_authorization: None,
-            platform_merchant_id,
+            platform_merchant_id: None,
             split_payments,
             force_3ds_challenge,
             force_3ds_challenge_trigger,
+            processor_merchant_id: Some(processor_merchant_id),
+            created_by: created_by.map(|cb| cb.to_string()),
         })
     }
     async fn convert_back(
@@ -1761,7 +1764,7 @@ impl behaviour::Conversion for PaymentIntent {
                 .transpose()
                 .change_context(common_utils::errors::CryptoError::DecodingFailed)?;
             Ok::<Self, error_stack::Report<common_utils::errors::CryptoError>>(Self {
-                merchant_id: storage_model.merchant_id,
+                merchant_id: storage_model.merchant_id.clone(),
                 status: storage_model.status,
                 amount_details,
                 amount_captured: storage_model.amount_captured,
@@ -1819,10 +1822,15 @@ impl behaviour::Conversion for PaymentIntent {
                 customer_present: storage_model.customer_present.into(),
                 payment_link_config: storage_model.payment_link_config,
                 routing_algorithm_id: storage_model.routing_algorithm_id,
-                platform_merchant_id: storage_model.platform_merchant_id,
                 split_payments: storage_model.split_payments,
                 force_3ds_challenge: storage_model.force_3ds_challenge,
                 force_3ds_challenge_trigger: storage_model.force_3ds_challenge_trigger,
+                processor_merchant_id: storage_model
+                    .processor_merchant_id
+                    .unwrap_or(storage_model.merchant_id),
+                created_by: storage_model
+                    .created_by
+                    .and_then(|created_by| created_by.parse::<CreatedBy>().ok()),
             })
         }
         .await
@@ -1903,9 +1911,11 @@ impl behaviour::Conversion for PaymentIntent {
             tax_details: amount_details.tax_details,
             enable_payment_link: Some(self.enable_payment_link.as_bool()),
             apply_mit_exemption: Some(self.apply_mit_exemption.as_bool()),
-            platform_merchant_id: self.platform_merchant_id,
+            platform_merchant_id: None,
             force_3ds_challenge: self.force_3ds_challenge,
             force_3ds_challenge_trigger: self.force_3ds_challenge_trigger,
+            processor_merchant_id: Some(self.processor_merchant_id),
+            created_by: self.created_by.map(|cb| cb.to_string()),
         })
     }
 }
@@ -1973,7 +1983,9 @@ impl behaviour::Conversion for PaymentIntent {
             skip_external_tax_calculation: self.skip_external_tax_calculation,
             request_extended_authorization: self.request_extended_authorization,
             psd2_sca_exemption_type: self.psd2_sca_exemption_type,
-            platform_merchant_id: self.platform_merchant_id,
+            platform_merchant_id: None,
+            processor_merchant_id: Some(self.processor_merchant_id),
+            created_by: self.created_by.map(|cb| cb.to_string()),
             force_3ds_challenge: self.force_3ds_challenge,
             force_3ds_challenge_trigger: self.force_3ds_challenge_trigger,
         })
@@ -2011,7 +2023,7 @@ impl behaviour::Conversion for PaymentIntent {
 
             Ok::<Self, error_stack::Report<common_utils::errors::CryptoError>>(Self {
                 payment_id: storage_model.payment_id,
-                merchant_id: storage_model.merchant_id,
+                merchant_id: storage_model.merchant_id.clone(),
                 status: storage_model.status,
                 amount: storage_model.amount,
                 currency: storage_model.currency,
@@ -2065,7 +2077,12 @@ impl behaviour::Conversion for PaymentIntent {
                 skip_external_tax_calculation: storage_model.skip_external_tax_calculation,
                 request_extended_authorization: storage_model.request_extended_authorization,
                 psd2_sca_exemption_type: storage_model.psd2_sca_exemption_type,
-                platform_merchant_id: storage_model.platform_merchant_id,
+                processor_merchant_id: storage_model
+                    .processor_merchant_id
+                    .unwrap_or(storage_model.merchant_id),
+                created_by: storage_model
+                    .created_by
+                    .and_then(|created_by| created_by.parse::<CreatedBy>().ok()),
                 force_3ds_challenge: storage_model.force_3ds_challenge,
                 force_3ds_challenge_trigger: storage_model.force_3ds_challenge_trigger,
             })
@@ -2133,7 +2150,9 @@ impl behaviour::Conversion for PaymentIntent {
             skip_external_tax_calculation: self.skip_external_tax_calculation,
             request_extended_authorization: self.request_extended_authorization,
             psd2_sca_exemption_type: self.psd2_sca_exemption_type,
-            platform_merchant_id: self.platform_merchant_id,
+            platform_merchant_id: None,
+            processor_merchant_id: Some(self.processor_merchant_id),
+            created_by: self.created_by.map(|cb| cb.to_string()),
             force_3ds_challenge: self.force_3ds_challenge,
             force_3ds_challenge_trigger: self.force_3ds_challenge_trigger,
         })
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 6953d1837d1..9d29c9ffaa4 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -3816,7 +3816,8 @@ mod tests {
             skip_external_tax_calculation: None,
             request_extended_authorization: None,
             psd2_sca_exemption_type: None,
-            platform_merchant_id: None,
+            processor_merchant_id: id_type::MerchantId::default(),
+            created_by: None,
             force_3ds_challenge: None,
             force_3ds_challenge_trigger: None,
         };
@@ -3890,7 +3891,8 @@ mod tests {
             skip_external_tax_calculation: None,
             request_extended_authorization: None,
             psd2_sca_exemption_type: None,
-            platform_merchant_id: None,
+            processor_merchant_id: id_type::MerchantId::default(),
+            created_by: None,
             force_3ds_challenge: None,
             force_3ds_challenge_trigger: None,
         };
@@ -3962,7 +3964,8 @@ mod tests {
             skip_external_tax_calculation: None,
             request_extended_authorization: None,
             psd2_sca_exemption_type: None,
-            platform_merchant_id: None,
+            processor_merchant_id: id_type::MerchantId::default(),
+            created_by: None,
             force_3ds_challenge: None,
             force_3ds_challenge_trigger: None,
         };
@@ -4500,6 +4503,8 @@ impl AttemptType {
             extended_authorization_applied: None,
             capture_before: None,
             card_discovery: None,
+            processor_merchant_id: old_payment_attempt.processor_merchant_id,
+            created_by: old_payment_attempt.created_by,
             setup_future_usage_applied: None,
         }
     }
diff --git a/crates/router/src/core/payments/operations/payment_approve.rs b/crates/router/src/core/payments/operations/payment_approve.rs
index c5837d5435b..377b930a572 100644
--- a/crates/router/src/core/payments/operations/payment_approve.rs
+++ b/crates/router/src/core/payments/operations/payment_approve.rs
@@ -11,7 +11,6 @@ use crate::{
     core::{
         errors::{self, RouterResult, StorageErrorExt},
         payments::{helpers, operations, PaymentData},
-        utils::ValidatePlatformMerchant,
     },
     events::audit_events::{AuditEvent, AuditEventType},
     routes::{app::ReqState, SessionState},
@@ -46,7 +45,7 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, api::PaymentsCaptureR
         key_store: &domain::MerchantKeyStore,
         _auth_flow: services::AuthFlow,
         _header_payload: &hyperswitch_domain_models::payments::HeaderPayload,
-        platform_merchant_account: Option<&domain::MerchantAccount>,
+        _platform_merchant_account: Option<&domain::MerchantAccount>,
     ) -> RouterResult<
         operations::GetTrackerResponse<'a, F, api::PaymentsCaptureRequest, PaymentData<F>>,
     > {
@@ -71,9 +70,6 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, api::PaymentsCaptureR
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         helpers::validate_payment_status_against_not_allowed_statuses(
             payment_intent.status,
             &[IntentStatus::Failed, IntentStatus::Succeeded],
diff --git a/crates/router/src/core/payments/operations/payment_cancel.rs b/crates/router/src/core/payments/operations/payment_cancel.rs
index 46ae753232f..9d4ff7a5579 100644
--- a/crates/router/src/core/payments/operations/payment_cancel.rs
+++ b/crates/router/src/core/payments/operations/payment_cancel.rs
@@ -12,7 +12,6 @@ use crate::{
     core::{
         errors::{self, RouterResult, StorageErrorExt},
         payments::{helpers, operations, PaymentData},
-        utils::ValidatePlatformMerchant,
     },
     events::audit_events::{AuditEvent, AuditEventType},
     routes::{app::ReqState, SessionState},
@@ -47,7 +46,7 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, api::PaymentsCancelRe
         key_store: &domain::MerchantKeyStore,
         _auth_flow: services::AuthFlow,
         _header_payload: &hyperswitch_domain_models::payments::HeaderPayload,
-        platform_merchant_account: Option<&domain::MerchantAccount>,
+        _platform_merchant_account: Option<&domain::MerchantAccount>,
     ) -> RouterResult<
         operations::GetTrackerResponse<'a, F, api::PaymentsCancelRequest, PaymentData<F>>,
     > {
@@ -71,9 +70,6 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, api::PaymentsCancelRe
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         helpers::validate_payment_status_against_not_allowed_statuses(
             payment_intent.status,
             &[
diff --git a/crates/router/src/core/payments/operations/payment_capture.rs b/crates/router/src/core/payments/operations/payment_capture.rs
index 2bde0d54976..68100cb6ece 100644
--- a/crates/router/src/core/payments/operations/payment_capture.rs
+++ b/crates/router/src/core/payments/operations/payment_capture.rs
@@ -11,7 +11,6 @@ use crate::{
     core::{
         errors::{self, RouterResult, StorageErrorExt},
         payments::{self, helpers, operations, types::MultipleCaptureData},
-        utils::ValidatePlatformMerchant,
     },
     events::audit_events::{AuditEvent, AuditEventType},
     routes::{app::ReqState, SessionState},
@@ -46,7 +45,7 @@ impl<F: Send + Clone + Sync> GetTracker<F, payments::PaymentData<F>, api::Paymen
         key_store: &domain::MerchantKeyStore,
         _auth_flow: services::AuthFlow,
         _header_payload: &hyperswitch_domain_models::payments::HeaderPayload,
-        platform_merchant_account: Option<&domain::MerchantAccount>,
+        _platform_merchant_account: Option<&domain::MerchantAccount>,
     ) -> RouterResult<
         operations::GetTrackerResponse<
             'a,
@@ -77,9 +76,6 @@ impl<F: Send + Clone + Sync> GetTracker<F, payments::PaymentData<F>, api::Paymen
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         payment_attempt = db
             .find_payment_attempt_by_payment_id_merchant_id_attempt_id(
                 &payment_intent.payment_id,
diff --git a/crates/router/src/core/payments/operations/payment_capture_v2.rs b/crates/router/src/core/payments/operations/payment_capture_v2.rs
index f69dd8de62c..4f9f9d31224 100644
--- a/crates/router/src/core/payments/operations/payment_capture_v2.rs
+++ b/crates/router/src/core/payments/operations/payment_capture_v2.rs
@@ -12,7 +12,6 @@ use crate::{
             helpers,
             operations::{self, ValidateStatusForOperation},
         },
-        utils::ValidatePlatformMerchant,
     },
     routes::{app::ReqState, SessionState},
     types::{
@@ -158,9 +157,6 @@ impl<F: Send + Clone> GetTracker<F, PaymentCaptureData<F>, PaymentsCaptureReques
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         self.validate_status_for_operation(payment_intent.status)?;
 
         let active_attempt_id = payment_intent
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index ef90a2f2bef..bc7d743e3f8 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -1366,6 +1366,8 @@ impl PaymentCreate {
                 extended_authorization_applied: None,
                 capture_before: None,
                 card_discovery: None,
+                processor_merchant_id: merchant_id.to_owned(),
+                created_by: None,
                 setup_future_usage_applied: request.setup_future_usage,
             },
             additional_pm_data,
@@ -1388,7 +1390,7 @@ impl PaymentCreate {
         active_attempt_id: String,
         profile_id: common_utils::id_type::ProfileId,
         session_expiry: PrimitiveDateTime,
-        platform_merchant_account: Option<&domain::MerchantAccount>,
+        _platform_merchant_account: Option<&domain::MerchantAccount>,
         business_profile: &domain::Profile,
     ) -> RouterResult<storage::PaymentIntent> {
         let created_at @ modified_at @ last_synced = common_utils::date_time::now();
@@ -1580,8 +1582,8 @@ impl PaymentCreate {
             skip_external_tax_calculation,
             request_extended_authorization: request.request_extended_authorization,
             psd2_sca_exemption_type: request.psd2_sca_exemption_type,
-            platform_merchant_id: platform_merchant_account
-                .map(|platform_merchant_account| platform_merchant_account.get_id().to_owned()),
+            processor_merchant_id: merchant_account.get_id().to_owned(),
+            created_by: None,
             force_3ds_challenge: request.force_3ds_challenge,
             force_3ds_challenge_trigger: Some(force_3ds_challenge_trigger),
         })
diff --git a/crates/router/src/core/payments/operations/payment_get.rs b/crates/router/src/core/payments/operations/payment_get.rs
index f15990d67b9..7e23cf6f84d 100644
--- a/crates/router/src/core/payments/operations/payment_get.rs
+++ b/crates/router/src/core/payments/operations/payment_get.rs
@@ -13,7 +13,6 @@ use crate::{
             helpers,
             operations::{self, ValidateStatusForOperation},
         },
-        utils::ValidatePlatformMerchant,
     },
     routes::{app::ReqState, SessionState},
     types::{
@@ -135,9 +134,6 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentStatusData<F>, PaymentsRetriev
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         let payment_attempt = payment_intent
             .active_attempt_id
             .as_ref()
diff --git a/crates/router/src/core/payments/operations/payment_get_intent.rs b/crates/router/src/core/payments/operations/payment_get_intent.rs
index 2e00b19c757..764fa937c23 100644
--- a/crates/router/src/core/payments/operations/payment_get_intent.rs
+++ b/crates/router/src/core/payments/operations/payment_get_intent.rs
@@ -10,7 +10,6 @@ use crate::{
     core::{
         errors::{self, RouterResult},
         payments::{self, helpers, operations},
-        utils::ValidatePlatformMerchant,
     },
     db::errors::StorageErrorExt,
     routes::{app::ReqState, SessionState},
@@ -100,9 +99,6 @@ impl<F: Send + Clone + Sync> GetTracker<F, payments::PaymentIntentData<F>, Payme
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         let payment_data = payments::PaymentIntentData {
             flow: PhantomData,
             payment_intent,
diff --git a/crates/router/src/core/payments/operations/payment_reject.rs b/crates/router/src/core/payments/operations/payment_reject.rs
index 82f09fc2075..0a230c3d110 100644
--- a/crates/router/src/core/payments/operations/payment_reject.rs
+++ b/crates/router/src/core/payments/operations/payment_reject.rs
@@ -11,7 +11,6 @@ use crate::{
     core::{
         errors::{self, RouterResult, StorageErrorExt},
         payments::{helpers, operations, PaymentAddress, PaymentData},
-        utils::ValidatePlatformMerchant,
     },
     events::audit_events::{AuditEvent, AuditEventType},
     routes::{app::ReqState, SessionState},
@@ -44,7 +43,7 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, PaymentsCancelRequest
         key_store: &domain::MerchantKeyStore,
         _auth_flow: services::AuthFlow,
         _header_payload: &hyperswitch_domain_models::payments::HeaderPayload,
-        platform_merchant_account: Option<&domain::MerchantAccount>,
+        _platform_merchant_account: Option<&domain::MerchantAccount>,
     ) -> RouterResult<operations::GetTrackerResponse<'a, F, PaymentsCancelRequest, PaymentData<F>>>
     {
         let db = &*state.store;
@@ -67,9 +66,6 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, PaymentsCancelRequest
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         helpers::validate_payment_status_against_not_allowed_statuses(
             payment_intent.status,
             &[
diff --git a/crates/router/src/core/payments/operations/payment_status.rs b/crates/router/src/core/payments/operations/payment_status.rs
index 5b5155bbe33..aab887b2c03 100644
--- a/crates/router/src/core/payments/operations/payment_status.rs
+++ b/crates/router/src/core/payments/operations/payment_status.rs
@@ -218,7 +218,7 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, api::PaymentsRetrieve
     ) -> RouterResult<
         operations::GetTrackerResponse<'a, F, api::PaymentsRetrieveRequest, PaymentData<F>>,
     > {
-        get_tracker_for_sync(
+        Box::pin(get_tracker_for_sync(
             payment_id,
             merchant_account,
             key_store,
@@ -227,7 +227,7 @@ impl<F: Send + Clone + Sync> GetTracker<F, PaymentData<F>, api::PaymentsRetrieve
             self,
             merchant_account.storage_scheme,
             platform_merchant_account,
-        )
+        ))
         .await
     }
 }
diff --git a/crates/router/src/core/payments/operations/payment_update_intent.rs b/crates/router/src/core/payments/operations/payment_update_intent.rs
index 5a283409edf..711115292ac 100644
--- a/crates/router/src/core/payments/operations/payment_update_intent.rs
+++ b/crates/router/src/core/payments/operations/payment_update_intent.rs
@@ -27,7 +27,6 @@ use crate::{
             self, helpers,
             operations::{self, ValidateStatusForOperation},
         },
-        utils::ValidatePlatformMerchant,
     },
     db::errors::StorageErrorExt,
     routes::{app::ReqState, SessionState},
@@ -150,9 +149,6 @@ impl<F: Send + Clone> GetTracker<F, payments::PaymentIntentData<F>, PaymentsUpda
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         self.validate_status_for_operation(payment_intent.status)?;
 
         let PaymentsUpdateIntentRequest {
diff --git a/crates/router/src/core/payments/operations/payments_incremental_authorization.rs b/crates/router/src/core/payments/operations/payments_incremental_authorization.rs
index dc87ca9140c..dfec91875a9 100644
--- a/crates/router/src/core/payments/operations/payments_incremental_authorization.rs
+++ b/crates/router/src/core/payments/operations/payments_incremental_authorization.rs
@@ -15,7 +15,6 @@ use crate::{
             self, helpers, operations, CustomerDetails, IncrementalAuthorizationDetails,
             PaymentAddress,
         },
-        utils::ValidatePlatformMerchant,
     },
     routes::{app::ReqState, SessionState},
     services,
@@ -49,7 +48,7 @@ impl<F: Send + Clone + Sync>
         key_store: &domain::MerchantKeyStore,
         _auth_flow: services::AuthFlow,
         _header_payload: &hyperswitch_domain_models::payments::HeaderPayload,
-        platform_merchant_account: Option<&domain::MerchantAccount>,
+        _platform_merchant_account: Option<&domain::MerchantAccount>,
     ) -> RouterResult<
         operations::GetTrackerResponse<
             'a,
@@ -78,9 +77,6 @@ impl<F: Send + Clone + Sync>
             .await
             .to_not_found_response(errors::ApiErrorResponse::PaymentNotFound)?;
 
-        payment_intent
-            .validate_platform_merchant(platform_merchant_account.map(|ma| ma.get_id()))?;
-
         helpers::validate_payment_status_against_allowed_statuses(
             payment_intent.status,
             &[enums::IntentStatus::RequiresCapture],
diff --git a/crates/router/src/core/payments/retry.rs b/crates/router/src/core/payments/retry.rs
index 9e179d55524..fb907a0f282 100644
--- a/crates/router/src/core/payments/retry.rs
+++ b/crates/router/src/core/payments/retry.rs
@@ -685,6 +685,8 @@ pub fn make_new_payment_attempt(
         extended_authorization_applied: Default::default(),
         capture_before: Default::default(),
         card_discovery: old_payment_attempt.card_discovery,
+        processor_merchant_id: old_payment_attempt.processor_merchant_id,
+        created_by: old_payment_attempt.created_by,
         setup_future_usage_applied: setup_futture_usage_intent, // setup future usage is picked from intent for new payment attempt
     }
 }
diff --git a/crates/router/src/core/utils.rs b/crates/router/src/core/utils.rs
index 87dce4ade0b..36e7423da18 100644
--- a/crates/router/src/core/utils.rs
+++ b/crates/router/src/core/utils.rs
@@ -1837,36 +1837,3 @@ pub(crate) fn validate_profile_id_from_auth_layer<T: GetProfileId + std::fmt::De
         (None, None) | (None, Some(_)) => Ok(()),
     }
 }
-
-pub(crate) trait ValidatePlatformMerchant {
-    fn get_platform_merchant_id(&self) -> Option<&common_utils::id_type::MerchantId>;
-
-    fn validate_platform_merchant(
-        &self,
-        auth_platform_merchant_id: Option<&common_utils::id_type::MerchantId>,
-    ) -> CustomResult<(), errors::ApiErrorResponse> {
-        let data_platform_merchant_id = self.get_platform_merchant_id();
-        match (data_platform_merchant_id, auth_platform_merchant_id) {
-            (Some(data_platform_merchant_id), Some(auth_platform_merchant_id)) => {
-                common_utils::fp_utils::when(
-                    data_platform_merchant_id != auth_platform_merchant_id,
-                    || {
-                        Err(report!(errors::ApiErrorResponse::PaymentNotFound)).attach_printable(format!(
-                     "Data platform merchant id: {data_platform_merchant_id:?} does not match with auth platform merchant id: {auth_platform_merchant_id:?}"))
-                    },
-                )
-            }
-            (Some(_), None) | (None, Some(_)) => {
-                Err(report!(errors::ApiErrorResponse::InvalidPlatformOperation))
-                    .attach_printable("Platform merchant id is missing in either data or auth")
-            }
-            (None, None) => Ok(()),
-        }
-    }
-}
-
-impl ValidatePlatformMerchant for storage::PaymentIntent {
-    fn get_platform_merchant_id(&self) -> Option<&common_utils::id_type::MerchantId> {
-        self.platform_merchant_id.as_ref()
-    }
-}
diff --git a/crates/router/src/types/storage/payment_attempt.rs b/crates/router/src/types/storage/payment_attempt.rs
index b2ef401abcf..3a942f70e32 100644
--- a/crates/router/src/types/storage/payment_attempt.rs
+++ b/crates/router/src/types/storage/payment_attempt.rs
@@ -222,6 +222,8 @@ mod tests {
             extended_authorization_applied: Default::default(),
             capture_before: Default::default(),
             card_discovery: Default::default(),
+            processor_merchant_id: Default::default(),
+            created_by: None,
             setup_future_usage_applied: Default::default(),
         };
 
@@ -310,6 +312,8 @@ mod tests {
             extended_authorization_applied: Default::default(),
             capture_before: Default::default(),
             card_discovery: Default::default(),
+            processor_merchant_id: Default::default(),
+            created_by: None,
             setup_future_usage_applied: Default::default(),
         };
         let store = state
@@ -411,6 +415,8 @@ mod tests {
             extended_authorization_applied: Default::default(),
             capture_before: Default::default(),
             card_discovery: Default::default(),
+            processor_merchant_id: Default::default(),
+            created_by: None,
             setup_future_usage_applied: Default::default(),
         };
         let store = state
diff --git a/crates/router/src/utils/user/sample_data.rs b/crates/router/src/utils/user/sample_data.rs
index 3ec06e4cd91..4dceb77d58d 100644
--- a/crates/router/src/utils/user/sample_data.rs
+++ b/crates/router/src/utils/user/sample_data.rs
@@ -276,7 +276,8 @@ pub async fn generate_sample_data(
             skip_external_tax_calculation: None,
             request_extended_authorization: None,
             psd2_sca_exemption_type: None,
-            platform_merchant_id: None,
+            processor_merchant_id: merchant_id.clone(),
+            created_by: None,
             force_3ds_challenge: None,
             force_3ds_challenge_trigger: None,
         };
@@ -368,6 +369,8 @@ pub async fn generate_sample_data(
             extended_authorization_applied: None,
             capture_before: None,
             card_discovery: None,
+            processor_merchant_id: Some(merchant_id.clone()),
+            created_by: None,
             setup_future_usage_applied: None,
         };
 
diff --git a/crates/storage_impl/src/mock_db/payment_attempt.rs b/crates/storage_impl/src/mock_db/payment_attempt.rs
index 1b97970bf16..5ce1c8b7b54 100644
--- a/crates/storage_impl/src/mock_db/payment_attempt.rs
+++ b/crates/storage_impl/src/mock_db/payment_attempt.rs
@@ -232,6 +232,8 @@ impl PaymentAttemptInterface for MockDb {
             charges: None,
             issuer_error_code: None,
             issuer_error_message: None,
+            processor_merchant_id: payment_attempt.processor_merchant_id,
+            created_by: payment_attempt.created_by,
             setup_future_usage_applied: payment_attempt.setup_future_usage_applied,
         };
         payment_attempts.push(payment_attempt.clone());
diff --git a/crates/storage_impl/src/payments/payment_attempt.rs b/crates/storage_impl/src/payments/payment_attempt.rs
index 25d89cbd6eb..98a7a62dc6b 100644
--- a/crates/storage_impl/src/payments/payment_attempt.rs
+++ b/crates/storage_impl/src/payments/payment_attempt.rs
@@ -3,7 +3,7 @@ use common_utils::types::keymanager::KeyManagerState;
 use common_utils::{
     errors::CustomResult,
     fallback_reverse_lookup_not_found,
-    types::{ConnectorTransactionId, ConnectorTransactionIdTrait},
+    types::{ConnectorTransactionId, ConnectorTransactionIdTrait, CreatedBy},
 };
 use diesel_models::{
     enums::{
@@ -650,6 +650,8 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
                     charges: None,
                     issuer_error_code: None,
                     issuer_error_message: None,
+                    processor_merchant_id: payment_attempt.processor_merchant_id.clone(),
+                    created_by: payment_attempt.created_by.clone(),
                     setup_future_usage_applied: payment_attempt.setup_future_usage_applied,
                 };
 
@@ -1657,6 +1659,8 @@ impl DataModelExt for PaymentAttempt {
             setup_future_usage_applied: self.setup_future_usage_applied,
             // Below fields are deprecated. Please add any new fields above this line.
             connector_transaction_data: None,
+            processor_merchant_id: Some(self.processor_merchant_id),
+            created_by: self.created_by.map(|created_by| created_by.to_string()),
         }
     }
 
@@ -1673,7 +1677,7 @@ impl DataModelExt for PaymentAttempt {
                 storage_model.tax_amount,
             ),
             payment_id: storage_model.payment_id,
-            merchant_id: storage_model.merchant_id,
+            merchant_id: storage_model.merchant_id.clone(),
             attempt_id: storage_model.attempt_id,
             status: storage_model.status,
             currency: storage_model.currency,
@@ -1740,6 +1744,12 @@ impl DataModelExt for PaymentAttempt {
             charges: storage_model.charges,
             issuer_error_code: storage_model.issuer_error_code,
             issuer_error_message: storage_model.issuer_error_message,
+            processor_merchant_id: storage_model
+                .processor_merchant_id
+                .unwrap_or(storage_model.merchant_id),
+            created_by: storage_model
+                .created_by
+                .and_then(|created_by| created_by.parse::<CreatedBy>().ok()),
             setup_future_usage_applied: storage_model.setup_future_usage_applied,
         }
     }
@@ -1827,6 +1837,8 @@ impl DataModelExt for PaymentAttemptNew {
             extended_authorization_applied: self.extended_authorization_applied,
             capture_before: self.capture_before,
             card_discovery: self.card_discovery,
+            processor_merchant_id: Some(self.processor_merchant_id),
+            created_by: self.created_by.map(|created_by| created_by.to_string()),
             setup_future_usage_applied: self.setup_future_usage_applied,
         }
     }
@@ -1841,7 +1853,7 @@ impl DataModelExt for PaymentAttemptNew {
                 storage_model.tax_amount,
             ),
             payment_id: storage_model.payment_id,
-            merchant_id: storage_model.merchant_id,
+            merchant_id: storage_model.merchant_id.clone(),
             attempt_id: storage_model.attempt_id,
             status: storage_model.status,
             currency: storage_model.currency,
@@ -1903,6 +1915,12 @@ impl DataModelExt for PaymentAttemptNew {
             extended_authorization_applied: storage_model.extended_authorization_applied,
             capture_before: storage_model.capture_before,
             card_discovery: storage_model.card_discovery,
+            processor_merchant_id: storage_model
+                .processor_merchant_id
+                .unwrap_or(storage_model.merchant_id),
+            created_by: storage_model
+                .created_by
+                .and_then(|created_by| created_by.parse::<CreatedBy>().ok()),
             setup_future_usage_applied: storage_model.setup_future_usage_applied,
         }
     }
diff --git a/migrations/2025-03-21-091838_add_new_col_processor_mid_and_created_by/down.sql b/migrations/2025-03-21-091838_add_new_col_processor_mid_and_created_by/down.sql
new file mode 100644
index 00000000000..f56ae9f5ef5
--- /dev/null
+++ b/migrations/2025-03-21-091838_add_new_col_processor_mid_and_created_by/down.sql
@@ -0,0 +1,5 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE payment_intent DROP COLUMN IF EXISTS processor_merchant_id;
+ALTER TABLE payment_intent DROP COLUMN IF EXISTS created_by;
+ALTER TABLE payment_attempt DROP COLUMN IF EXISTS processor_merchant_id;
+ALTER TABLE payment_attempt DROP COLUMN IF EXISTS created_by;
diff --git a/migrations/2025-03-21-091838_add_new_col_processor_mid_and_created_by/up.sql b/migrations/2025-03-21-091838_add_new_col_processor_mid_and_created_by/up.sql
new file mode 100644
index 00000000000..52932564b4b
--- /dev/null
+++ b/migrations/2025-03-21-091838_add_new_col_processor_mid_and_created_by/up.sql
@@ -0,0 +1,8 @@
+-- Your SQL goes here
+ALTER TABLE payment_intent ADD COLUMN IF NOT EXISTS processor_merchant_id VARCHAR(64);
+ALTER TABLE payment_intent ADD COLUMN IF NOT EXISTS created_by VARCHAR(255);
+ALTER TABLE payment_attempt ADD COLUMN IF NOT EXISTS processor_merchant_id VARCHAR(64);
+ALTER TABLE payment_attempt ADD COLUMN IF NOT EXISTS created_by VARCHAR(255);
+-- This backfill should be executed again after deployment.
+UPDATE payment_intent SET processor_merchant_id = merchant_id where processor_merchant_id IS NULL;
+UPDATE payment_attempt SET processor_merchant_id = merchant_id where processor_merchant_id IS NULL;

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

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

## Description
<!-- Describe your changes in detail -->
- added `processor_merchant_id` and` created_by` column in `diesel_models` and `hyperswitch_domain_models`
- added migrations
- ~added backfilling `processor_merchant_id` column in migrations~ (Due to staggered release backfilling should be done at a later point after deployment is complete so it can't be done in the scope of this PR)
- populating these fields with default scenario values as mentions in [Motivation and Context](#motivation-and-context)
- removing `platform_merchant_id` column and population instances from `hyperswitch_domain_models`.
- created a macro `impl_enum_str!` to implement serialization and deserialization for any enum with singleton associated data.
  - for e.g 
  ```rust 
  CreatedBy::Api { merchant_id: merchant_32423 } 
  ```
   will be serialized to `Api:merchant_32423`

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

Fixes #7769 

#### Introducing two fields in payment_intent and payment_attempt table namely, 
  - `processor_merchant_id`
  - `created_by`
  
#### Terminology
| column | description |
| ----------| -------- |
| **merchant_id**(already exists) | Who actually configures payment settings, or who has integrated with us, i.e **owner**. |
| **processor_merchant_id** | Who owns the credentials of the processor, i.e. processor owner |
| **created_by** | who invoked the resource based api (identifier) and through what source (Api, Jwt(Dashboard)) |

#### Value combination matrix
 | Cases                                               | merchant_id | processor_merchant_id | created_by(null) |
| --------------------------------------------------- | ----------- | --------------------- | -----------|
| Normal (non vsas)                                   | 123         | 123                   | 123              | 
| Platform performing for connected                   | pid1        | 123                   | pid1             |
| Platform performing for platform                    | pid1        | pid1                  | pid1             |
| Connected making some operation on top of a payment | pid1        | 123                   | 123              | 
>created_by is an enum struct
```rust
pub enum CreatedBy {
    Api { merchant_id: id_type::MerchantId },
    Jwt { user_id: String },
}
```

#### **Important Note**: This PR is just a scaffold for the next changes to come that will include several logic like, determining what scenario it is mentioned in the above table and populating column fields accordingly.

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

### Steps
1. Make a payment
```sh
curl --location '{{baseurl}}/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: ****redacted****' \
--data-raw '{
    "amount": 104000,
    "currency": "GBP",
    "amount_to_capture": 104000,
    "confirm": true,
    "profile_id": "**redacted**",
    "capture_method": "automatic",
    "capture_on": "2022-09-10T10:11:12Z",
    "authentication_type": "no_three_ds",
    "setup_future_usage": "on_session", 
    "customer": {
        "id": "customer123",
        "name": "John Doe",
        "email": "customer@gmail.com",
        "phone": "9999999999",
        "phone_country_code": "+1"
    },
    "customer_id": "customer123",
    
    
    
    "phone_country_code": "+1",
    "routing": { 
        "type": "single",
        "data": "stripe"
    },
    "description": "Its my first payment request",
    "return_url": "https://google.com",
    "payment_method": "card",
    "payment_method_type": "credit",
    "payment_method_data": { 
        "card": {
            "card_number": "re",
            "card_exp_month": "01",
            "card_exp_year": "99",
            "card_holder_name": "John Smith",
            "card_cvc": "257"
        }
    },
    
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "8056594427",
            "country_code": "+91"
        },
        "email": "guest@example.com"
    },
    "shipping": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "8056594427",
            "country_code": "+91"
        },
        "email": "guest@example.com"
    },
    "statement_descriptor_name": "joseph",
    "statement_descriptor_suffix": "JS",
    "order_details": [
        {
            "product_name": "Apple iphone 15",
            "quantity": 1,
            "amount": 104000,
            "account_name": "transaction_processing"
        }
    ],
    "metadata": {
        "udf1": "value1",
        "new_customer": "true",
        "login_date": "2019-09-10T10:11:12Z"
    },
    "browser_info": {
        "user_agent": "Mozilla\/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit\/537.36 (KHTML, like Gecko) Chrome\/70.0.3538.110 Safari\/537.36",
        "accept_header": "text\/html,application\/xhtml+xml,application\/xml;q=0.9,image\/webp,image\/apng,*\/*;q=0.8",
        "language": "nl-NL",
        "color_depth": 24,
        "screen_height": 723,
        "screen_width": 1536,
        "time_zone": 0,
        "java_enabled": true,
        "java_script_enabled": true,
        "ip_address": "128.0.0.1"
    },
    "customer_acceptance": {
        "acceptance_type": "offline",
        "accepted_at": "1963-05-03T04:07:52.723Z",
        "online": {
            "ip_address": "125.0.0.1",
            "user_agent": "amet irure esse"
        }
    },
    
    "connector_metadata": {
        "noon": {
            "order_category": "pay"
        }
    },
    "payment_link": false,
    "payment_link_config": {
        "theme": "",
        "logo": "",
        "seller_name": "",
        "sdk_layout": "",
        "display_sdk_only": false,
        "enabled_saved_payment_method": false
    },
    "payment_type": "normal", 
    "request_incremental_authorization": false,
    "merchant_order_reference_id": "test_ord",
    "session_expiry": 900
}'
```
response 
```json
{
    "payment_id": "pay_mzSYO8ZS6YgxJ8qhpk10",
    "merchant_id": "merchant_1744876813",
    "status": "processing",
    "amount": 104000,
    "net_amount": 104000,
    "shipping_cost": null,
    "amount_capturable": 0,
    "amount_received": null,
    "connector": "checkout",
    "client_secret": "****redacted****",
    "created": "2025-04-17T08:11:05.286Z",
    "currency": "GBP",
    "customer_id": "customer123",
    "customer": {
        "id": "customer123",
        "name": "John Doe",
        "email": "customer@gmail.com",
        "phone": "9999999999",
        "phone_country_code": "+1"
    },
    "description": "Its my first payment request",
    "refunds": null,
    "disputes": null,
    "mandate_id": null,
    "mandate_data": null,
    "setup_future_usage": "on_session",
    "off_session": null,
    "capture_on": null,
    "capture_method": "automatic",
    "payment_method": "card",
    "payment_method_data": {
        "card": {
            "last4": "0131",
            "card_type": null,
            "card_network": null,
            "card_issuer": null,
            "card_issuing_country": null,
            "card_isin": "453243",
            "card_extended_bin": null,
            "card_exp_month": "01",
            "card_exp_year": "99",
            "card_holder_name": "John Smith",
            "payment_checks": null,
            "authentication_data": null
        },
        "billing": null
    },
    "payment_token": null,
    "shipping": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "8056594427",
            "country_code": "+91"
        },
        "email": "guest@example.com"
    },
    "billing": {
        "address": {
            "city": "San Fransico",
            "country": "US",
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "zip": "94122",
            "state": "California",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "8056594427",
            "country_code": "+91"
        },
        "email": "guest@example.com"
    },
    "order_details": [
        {
            "brand": null,
            "amount": 104000,
            "category": null,
            "quantity": 1,
            "tax_rate": null,
            "product_id": null,
            "product_name": "Apple iphone 15",
            "product_type": null,
            "sub_category": null,
            "product_img_link": null,
            "product_tax_code": null,
            "total_tax_amount": null,
            "requires_shipping": null
        }
    ],
    "email": "customer@gmail.com",
    "name": "John Doe",
    "phone": "9999999999",
    "return_url": "https://google.com/",
    "authentication_type": "no_three_ds",
    "statement_descriptor_name": "joseph",
    "statement_descriptor_suffix": "JS",
    "next_action": null,
    "cancellation_reason": null,
    "error_code": null,
    "error_message": null,
    "unified_code": null,
    "unified_message": null,
    "payment_experience": null,
    "payment_method_type": "credit",
    "connector_label": null,
    "business_country": null,
    "business_label": "default",
    "business_sub_label": null,
    "allowed_payment_method_types": null,
    "ephemeral_key": {
        "customer_id": "customer123",
        "created_at": 1744877465,
        "expires": 1744881065,
        "secret": "****redacted****"
    },
    "manual_retry_allowed": false,
    "connector_transaction_id": "pay_smcaidph7qhefdvm653lrn6wda",
    "frm_message": null,
    "metadata": {
        "udf1": "value1",
        "login_date": "2019-09-10T10:11:12Z",
        "new_customer": "true"
    },
    "connector_metadata": {
        "apple_pay": null,
        "airwallex": null,
        "noon": {
            "order_category": "pay"
        },
        "braintree": null
    },
    "feature_metadata": null,
    "reference_id": "pay_mzSYO8ZS6YgxJ8qhpk10_1",
    "payment_link": null,
    "profile_id": "pro_RZojFbEiPLcW3fj9rK1W",
    "surcharge_details": null,
    "attempt_count": 1,
    "merchant_decision": null,
    "merchant_connector_id": "mca_J1wam8mC5wvSWgcWAoiz",
    "incremental_authorization_allowed": false,
    "authorization_count": null,
    "incremental_authorizations": null,
    "external_authentication_details": null,
    "external_3ds_authentication_attempted": false,
    "expires_on": "2025-04-17T08:26:05.286Z",
    "fingerprint": null,
    "browser_info": {
        "language": "nl-NL",
        "time_zone": 0,
        "ip_address": "128.0.0.1",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36",
        "color_depth": 24,
        "java_enabled": true,
        "screen_width": 1536,
        "accept_header": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
        "screen_height": 723,
        "java_script_enabled": true
    },
    "payment_method_id": null,
    "payment_method_status": null,
    "updated": "2025-04-17T08:11:06.465Z",
    "split_payments": null,
    "frm_metadata": null,
    "extended_authorization_applied": null,
    "capture_before": null,
    "merchant_order_reference_id": "test_ord",
    "order_tax_amount": null,
    "connector_mandate_id": null,
    "card_discovery": "manual",
    "force_3ds_challenge": false,
    "force_3ds_challenge_trigger": false,
    "issuer_error_code": null,
    "issuer_error_message": null
}
```
2.  run the sql query in psql console
```sql
select merchant_id, processor_merchant_id, created_by from payment_intent;
```
run the above query
### Result 

| merchant_id     | processor_merchant_id |       created_by|
| --------------------- | ----------------------- | ------------------------- |
| merchant_1 | merchant_1  |            |
| merchant_2 | merchant_2  |          |

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [x] I added unit tests for my changes where possible
