Issue:
Bug: Add profile id to transaction tables



Required to provide profile level filters in analytics section

Pull request description:
## 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 -->
Add organisation id for 
- payment intents
- payment attempts
- refunds
- disputes
add profile id & card network for payment attempts

Add the above fields in clickhouse & kafka as well

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] 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).
-->

Adding these fields to provide better authentication guards

## 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)?
-->
<img width="975" alt="image" src="https://github.com/user-attachments/assets/dcb1ff45-4771-4382-9f97-36039abb9ef2">

Check for the added columns in pgweb/grafana

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [ ] I formatted the code `cargo +nightly fmt --all`
- [ ] I addressed lints thrown by `cargo clippy`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible


Testing:

<img width="975" alt="image" src="https://github.com/user-attachments/assets/dcb1ff45-4771-4382-9f97-36039abb9ef2">

Check for the added columns in pgweb/grafana

Patch:
diff --git a/crates/analytics/docs/clickhouse/scripts/disputes.sql b/crates/analytics/docs/clickhouse/scripts/disputes.sql
index bb7472a4d54..3bd993d31cf 100644
--- a/crates/analytics/docs/clickhouse/scripts/disputes.sql
+++ b/crates/analytics/docs/clickhouse/scripts/disputes.sql
@@ -20,6 +20,7 @@ CREATE TABLE dispute_queue (
     `evidence` Nullable(String),
     `profile_id` Nullable(String),
     `merchant_connector_id` Nullable(String),
+    `organization_id` String,
     `sign_flag` Int8
 ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
 kafka_topic_list = 'hyperswitch-dispute-events',
@@ -50,6 +51,7 @@ CREATE TABLE dispute (
     `profile_id` Nullable(String),
     `merchant_connector_id` Nullable(String),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
+    `organization_id` String,
     `sign_flag` Int8,
     INDEX connectorIndex connector TYPE bloom_filter GRANULARITY 1,
     INDEX disputeStatusIndex dispute_status TYPE bloom_filter GRANULARITY 1,
@@ -80,6 +82,7 @@ CREATE MATERIALIZED VIEW dispute_mv TO dispute (
     `evidence` Nullable(String),
     `profile_id` Nullable(String),
     `merchant_connector_id` Nullable(String),
+    `organization_id` String,
     `inserted_at` DateTime64(3),
     `sign_flag` Int8
 ) AS
@@ -105,6 +108,7 @@ SELECT
     evidence,
     profile_id,
     merchant_connector_id,
+    organization_id,
     now() AS inserted_at,
     sign_flag
 FROM
diff --git a/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql b/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql
index 223d29c9836..5c1ee8754c9 100644
--- a/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql
+++ b/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql
@@ -40,6 +40,8 @@ CREATE TABLE payment_attempt_queue (
     `mandate_data` Nullable(String),
     `client_source` LowCardinality(Nullable(String)),
     `client_version` LowCardinality(Nullable(String)),
+    `organization_id` String,
+    `profile_id` String,
     `sign_flag` Int8
 ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
 kafka_topic_list = 'hyperswitch-payment-attempt-events',
@@ -90,6 +92,8 @@ CREATE TABLE payment_attempts (
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
     `client_source` LowCardinality(Nullable(String)),
     `client_version` LowCardinality(Nullable(String)),
+    `organization_id` String,
+    `profile_id` String,
     `sign_flag` Int8,
     INDEX connectorIndex connector TYPE bloom_filter GRANULARITY 1,
     INDEX paymentMethodIndex payment_method TYPE bloom_filter GRANULARITY 1,
@@ -143,6 +147,8 @@ CREATE MATERIALIZED VIEW payment_attempt_mv TO payment_attempts (
     `inserted_at` DateTime64(3),
     `client_source` LowCardinality(Nullable(String)),
     `client_version` LowCardinality(Nullable(String)),
+    `organization_id` String,
+    `profile_id` String,
     `sign_flag` Int8
 ) AS
 SELECT
@@ -188,6 +194,8 @@ SELECT
     now() AS inserted_at,
     client_source,
     client_version,
+    organization_id,
+    profile_id,
     sign_flag
 FROM
     payment_attempt_queue
diff --git a/crates/analytics/docs/clickhouse/scripts/payment_intents.sql b/crates/analytics/docs/clickhouse/scripts/payment_intents.sql
index 09101e9aa2d..b4d00a858b0 100644
--- a/crates/analytics/docs/clickhouse/scripts/payment_intents.sql
+++ b/crates/analytics/docs/clickhouse/scripts/payment_intents.sql
@@ -23,6 +23,7 @@ CREATE TABLE payment_intents_queue
     `modified_at` DateTime CODEC(T64, LZ4),
     `created_at` DateTime CODEC(T64, LZ4),
     `last_synced` Nullable(DateTime) CODEC(T64, LZ4),
+    `organization_id` String,
     `sign_flag` Int8
 ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
 kafka_topic_list = 'hyperswitch-payment-intent-events',
@@ -56,6 +57,7 @@ CREATE TABLE payment_intents
     `created_at` DateTime DEFAULT now() CODEC(T64, LZ4),
     `last_synced` Nullable(DateTime) CODEC(T64, LZ4),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
+    `organization_id` String,
     `sign_flag` Int8,
     INDEX connectorIndex connector_id TYPE bloom_filter GRANULARITY 1,
     INDEX currencyIndex currency TYPE bloom_filter GRANULARITY 1,
@@ -93,6 +95,7 @@ CREATE MATERIALIZED VIEW payment_intents_mv TO payment_intents
     `created_at` DateTime64(3),
     `last_synced` Nullable(DateTime64(3)),
     `inserted_at` DateTime64(3),
+    `organization_id` String,
     `sign_flag` Int8
 ) AS
 SELECT
@@ -120,5 +123,6 @@ SELECT
     created_at,
     last_synced,
     now() AS inserted_at,
+    organization_id,
     sign_flag
 FROM payment_intents_queue;
\ No newline at end of file
diff --git a/crates/analytics/docs/clickhouse/scripts/refunds.sql b/crates/analytics/docs/clickhouse/scripts/refunds.sql
index 74c069db853..323caf1adb4 100644
--- a/crates/analytics/docs/clickhouse/scripts/refunds.sql
+++ b/crates/analytics/docs/clickhouse/scripts/refunds.sql
@@ -21,6 +21,8 @@ CREATE TABLE refund_queue (
     `refund_error_code` Nullable(String),
     `created_at` DateTime,
     `modified_at` DateTime,
+    `organization_id` String,
+    `profile_id` String,
     `sign_flag` Int8
 ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
 kafka_topic_list = 'hyperswitch-refund-events',
@@ -52,6 +54,8 @@ CREATE TABLE refunds (
     `created_at` DateTime DEFAULT now() CODEC(T64, LZ4),
     `modified_at` DateTime DEFAULT now() CODEC(T64, LZ4),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
+    `organization_id` String,
+    `profile_id` String,
     `sign_flag` Int8,
     INDEX connectorIndex connector TYPE bloom_filter GRANULARITY 1,
     INDEX refundTypeIndex refund_type TYPE bloom_filter GRANULARITY 1,
@@ -85,6 +89,8 @@ CREATE MATERIALIZED VIEW refund_mv TO refunds (
     `created_at` DateTime64(3),
     `modified_at` DateTime64(3),
     `inserted_at` DateTime64(3),
+    `organization_id` String,
+    `profile_id` String,
     `sign_flag` Int8
 ) AS
 SELECT
@@ -111,6 +117,8 @@ SELECT
     created_at,
     modified_at,
     now() AS inserted_at,
+    organization_id,
+    profile_id,
     sign_flag
 FROM
     refund_queue
diff --git a/crates/diesel_models/src/customers.rs b/crates/diesel_models/src/customers.rs
index 7e4e27a5d38..c0d6639d1ad 100644
--- a/crates/diesel_models/src/customers.rs
+++ b/crates/diesel_models/src/customers.rs
@@ -41,6 +41,7 @@ impl Customer {
 
 #[cfg(all(feature = "v2", feature = "customer_v2"))]
 impl Customer {
+    #[allow(clippy::todo)]
     pub fn get_customer_id(&self) -> common_utils::id_type::CustomerId {
         todo!()
     }
diff --git a/crates/diesel_models/src/dispute.rs b/crates/diesel_models/src/dispute.rs
index 45004e10ba5..1c70655406a 100644
--- a/crates/diesel_models/src/dispute.rs
+++ b/crates/diesel_models/src/dispute.rs
@@ -30,6 +30,7 @@ pub struct DisputeNew {
     pub profile_id: Option<common_utils::id_type::ProfileId>,
     pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
     pub dispute_amount: i64,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Identifiable, Queryable, Selectable)]
@@ -59,6 +60,7 @@ pub struct Dispute {
     pub profile_id: Option<common_utils::id_type::ProfileId>,
     pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
     pub dispute_amount: i64,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[derive(Debug)]
diff --git a/crates/diesel_models/src/payment_attempt.rs b/crates/diesel_models/src/payment_attempt.rs
index ec87b2da2e4..fd769d4dbd2 100644
--- a/crates/diesel_models/src/payment_attempt.rs
+++ b/crates/diesel_models/src/payment_attempt.rs
@@ -1,4 +1,4 @@
-use common_utils::pii;
+use common_utils::{id_type, pii};
 use diesel::{AsChangeset, Identifiable, Insertable, Queryable, Selectable};
 use serde::{Deserialize, Serialize};
 use time::PrimitiveDateTime;
@@ -16,7 +16,7 @@ use crate::schema_v2::payment_attempt;
 #[diesel(table_name = payment_attempt, primary_key(attempt_id, merchant_id), check_for_backend(diesel::pg::Pg))]
 pub struct PaymentAttempt {
     pub payment_id: String,
-    pub merchant_id: common_utils::id_type::MerchantId,
+    pub merchant_id: id_type::MerchantId,
     pub attempt_id: String,
     pub status: storage_enums::AttemptStatus,
     pub amount: i64,
@@ -62,7 +62,7 @@ pub struct PaymentAttempt {
     pub connector_response_reference_id: Option<String>,
     pub amount_capturable: i64,
     pub updated_by: String,
-    pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+    pub merchant_connector_id: Option<id_type::MerchantConnectorAccountId>,
     pub authentication_data: Option<serde_json::Value>,
     pub encoded_data: Option<String>,
     pub unified_code: Option<String>,
@@ -78,6 +78,9 @@ pub struct PaymentAttempt {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub profile_id: id_type::ProfileId,
+    pub organization_id: id_type::OrganizationId,
+    pub card_network: Option<String>,
 }
 
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "payment_v2")))]
@@ -87,7 +90,7 @@ pub struct PaymentAttempt {
 #[diesel(table_name = payment_attempt, primary_key(attempt_id, merchant_id), check_for_backend(diesel::pg::Pg))]
 pub struct PaymentAttempt {
     pub payment_id: String,
-    pub merchant_id: common_utils::id_type::MerchantId,
+    pub merchant_id: id_type::MerchantId,
     pub attempt_id: String,
     pub status: storage_enums::AttemptStatus,
     pub amount: i64,
@@ -133,7 +136,7 @@ pub struct PaymentAttempt {
     pub connector_response_reference_id: Option<String>,
     pub amount_capturable: i64,
     pub updated_by: String,
-    pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+    pub merchant_connector_id: Option<id_type::MerchantConnectorAccountId>,
     pub authentication_data: Option<serde_json::Value>,
     pub encoded_data: Option<String>,
     pub unified_code: Option<String>,
@@ -149,6 +152,9 @@ pub struct PaymentAttempt {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub profile_id: id_type::ProfileId,
+    pub organization_id: id_type::OrganizationId,
+    pub card_network: Option<String>,
 }
 
 impl PaymentAttempt {
@@ -171,7 +177,7 @@ pub struct PaymentListFilters {
 #[diesel(table_name = payment_attempt)]
 pub struct PaymentAttemptNew {
     pub payment_id: String,
-    pub merchant_id: common_utils::id_type::MerchantId,
+    pub merchant_id: id_type::MerchantId,
     pub attempt_id: String,
     pub status: storage_enums::AttemptStatus,
     pub amount: i64,
@@ -215,7 +221,7 @@ pub struct PaymentAttemptNew {
     pub multiple_capture_count: Option<i16>,
     pub amount_capturable: i64,
     pub updated_by: String,
-    pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+    pub merchant_connector_id: Option<id_type::MerchantConnectorAccountId>,
     pub authentication_data: Option<serde_json::Value>,
     pub encoded_data: Option<String>,
     pub unified_code: Option<String>,
@@ -231,6 +237,9 @@ pub struct PaymentAttemptNew {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub profile_id: id_type::ProfileId,
+    pub organization_id: id_type::OrganizationId,
+    pub card_network: Option<String>,
 }
 
 impl PaymentAttemptNew {
@@ -280,7 +289,7 @@ pub enum PaymentAttemptUpdate {
         surcharge_amount: Option<i64>,
         tax_amount: Option<i64>,
         updated_by: String,
-        merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+        merchant_connector_id: Option<id_type::MerchantConnectorAccountId>,
     },
     AuthenticationTypeUpdate {
         authentication_type: storage_enums::AuthenticationType,
@@ -308,7 +317,7 @@ pub enum PaymentAttemptUpdate {
         tax_amount: Option<i64>,
         fingerprint_id: Option<String>,
         updated_by: String,
-        merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
+        merchant_connector_id: Option<id_type::MerchantConnectorAccountId>,
         payment_method_id: Option<String>,
         external_three_ds_authentication_attempted: Option<bool>,
         authentication_connector: Option<String>,
@@ -475,7 +484,7 @@ pub struct PaymentAttemptUpdateInternal {
     tax_amount: Option<i64>,
     amount_capturable: Option<i64>,
     updated_by: String,
-    merchant_connector_id: Option<Option<common_utils::id_type::MerchantConnectorAccountId>>,
+    merchant_connector_id: Option<Option<id_type::MerchantConnectorAccountId>>,
     authentication_data: Option<serde_json::Value>,
     encoded_data: Option<String>,
     unified_code: Option<Option<String>>,
@@ -489,6 +498,7 @@ pub struct PaymentAttemptUpdateInternal {
     client_source: Option<String>,
     client_version: Option<String>,
     customer_acceptance: Option<pii::SecretSerdeValue>,
+    card_network: Option<String>,
 }
 
 impl PaymentAttemptUpdateInternal {
@@ -505,6 +515,15 @@ impl PaymentAttemptUpdateInternal {
                     .or(source.tax_amount)
                     .unwrap_or(0),
         );
+        update_internal.card_network = update_internal
+            .payment_method_data
+            .as_ref()
+            .and_then(|data| data.as_object())
+            .and_then(|card| card.get("card"))
+            .and_then(|data| data.as_object())
+            .and_then(|card| card.get("card_network"))
+            .and_then(|network| network.as_str())
+            .map(|network| network.to_string());
         update_internal
     }
 }
@@ -558,6 +577,7 @@ impl PaymentAttemptUpdate {
             client_source,
             client_version,
             customer_acceptance,
+            card_network,
         } = PaymentAttemptUpdateInternal::from(self).populate_derived_fields(&source);
         PaymentAttempt {
             amount: amount.unwrap_or(source.amount),
@@ -610,6 +630,7 @@ impl PaymentAttemptUpdate {
             client_source: client_source.or(source.client_source),
             client_version: client_version.or(source.client_version),
             customer_acceptance: customer_acceptance.or(source.customer_acceptance),
+            card_network: card_network.or(source.card_network),
             ..source
         }
     }
@@ -685,6 +706,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::AuthenticationTypeUpdate {
                 authentication_type,
@@ -736,6 +758,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::ConfirmUpdate {
                 amount,
@@ -815,6 +838,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 unified_code: None,
                 unified_message: None,
                 charge_id: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::VoidUpdate {
                 status,
@@ -867,6 +891,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::RejectUpdate {
                 status,
@@ -920,6 +945,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::BlocklistUpdate {
                 status,
@@ -973,6 +999,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::PaymentMethodDetailsUpdate {
                 payment_method_id,
@@ -1024,6 +1051,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::ResponseUpdate {
                 status,
@@ -1093,6 +1121,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::ErrorUpdate {
                 connector,
@@ -1154,6 +1183,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::StatusUpdate { status, updated_by } => Self {
                 status: Some(status),
@@ -1202,6 +1232,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::UpdateTrackers {
                 payment_token,
@@ -1259,6 +1290,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::UnresolvedResponseUpdate {
                 status,
@@ -1317,6 +1349,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::PreprocessingUpdate {
                 status,
@@ -1373,6 +1406,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::CaptureUpdate {
                 multiple_capture_count,
@@ -1425,6 +1459,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::AmountToCaptureUpdate {
                 status,
@@ -1477,6 +1512,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::ConnectorResponse {
                 authentication_data,
@@ -1532,6 +1568,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::IncrementalAuthorizationAmountUpdate {
                 amount,
@@ -1583,6 +1620,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::AuthenticationUpdate {
                 status,
@@ -1637,6 +1675,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
             PaymentAttemptUpdate::ManualUpdate {
                 status,
@@ -1694,6 +1733,7 @@ impl From<PaymentAttemptUpdate> for PaymentAttemptUpdateInternal {
                 client_source: None,
                 client_version: None,
                 customer_acceptance: None,
+                card_network: None,
             },
         }
     }
diff --git a/crates/diesel_models/src/payment_intent.rs b/crates/diesel_models/src/payment_intent.rs
index 7bd1652ab56..c96629ea448 100644
--- a/crates/diesel_models/src/payment_intent.rs
+++ b/crates/diesel_models/src/payment_intent.rs
@@ -69,6 +69,7 @@ pub struct PaymentIntent {
     pub merchant_order_reference_id: Option<String>,
     pub shipping_details: Option<Encryption>,
     pub is_payment_processor_token_flow: Option<bool>,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[cfg(all(any(feature = "v1", feature = "v2"), not(feature = "payment_v2")))]
@@ -130,6 +131,7 @@ pub struct PaymentIntent {
     pub merchant_order_reference_id: Option<String>,
     pub shipping_details: Option<Encryption>,
     pub is_payment_processor_token_flow: Option<bool>,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[derive(
@@ -190,6 +192,7 @@ pub struct PaymentIntentNew {
     pub merchant_order_reference_id: Option<String>,
     pub shipping_details: Option<Encryption>,
     pub is_payment_processor_token_flow: Option<bool>,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/crates/diesel_models/src/refund.rs b/crates/diesel_models/src/refund.rs
index 10c1fbb9b29..a0828b31756 100644
--- a/crates/diesel_models/src/refund.rs
+++ b/crates/diesel_models/src/refund.rs
@@ -50,6 +50,7 @@ pub struct Refund {
     pub updated_by: String,
     pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
     pub charges: Option<ChargeRefunds>,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[derive(
@@ -92,38 +93,7 @@ pub struct RefundNew {
     pub updated_by: String,
     pub merchant_connector_id: Option<common_utils::id_type::MerchantConnectorAccountId>,
     pub charges: Option<ChargeRefunds>,
-}
-
-impl Default for RefundNew {
-    fn default() -> Self {
-        Self {
-            refund_id: Default::default(),
-            payment_id: Default::default(),
-            merchant_id: Default::default(),
-            internal_reference_id: Default::default(),
-            external_reference_id: Default::default(),
-            connector_transaction_id: Default::default(),
-            connector: Default::default(),
-            connector_refund_id: Default::default(),
-            refund_type: Default::default(),
-            total_amount: Default::default(),
-            currency: Default::default(),
-            refund_amount: Default::default(),
-            refund_status: Default::default(),
-            sent_to_gateway: Default::default(),
-            metadata: Default::default(),
-            refund_arn: Default::default(),
-            created_at: common_utils::date_time::now(),
-            modified_at: common_utils::date_time::now(),
-            description: Default::default(),
-            attempt_id: Default::default(),
-            refund_reason: Default::default(),
-            profile_id: Default::default(),
-            updated_by: Default::default(),
-            merchant_connector_id: Default::default(),
-            charges: Default::default(),
-        }
-    }
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
diff --git a/crates/diesel_models/src/schema.rs b/crates/diesel_models/src/schema.rs
index 7e6c1b84889..3e4799e42ab 100644
--- a/crates/diesel_models/src/schema.rs
+++ b/crates/diesel_models/src/schema.rs
@@ -375,6 +375,8 @@ diesel::table! {
         #[max_length = 32]
         merchant_connector_id -> Nullable<Varchar>,
         dispute_amount -> Int8,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 
@@ -831,6 +833,12 @@ diesel::table! {
         #[max_length = 64]
         client_version -> Nullable<Varchar>,
         customer_acceptance -> Nullable<Jsonb>,
+        #[max_length = 64]
+        profile_id -> Varchar,
+        #[max_length = 32]
+        organization_id -> Varchar,
+        #[max_length = 32]
+        card_network -> Nullable<Varchar>,
     }
 }
 
@@ -906,6 +914,8 @@ diesel::table! {
         merchant_order_reference_id -> Nullable<Varchar>,
         shipping_details -> Nullable<Bytea>,
         is_payment_processor_token_flow -> Nullable<Bool>,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 
@@ -1150,6 +1160,8 @@ diesel::table! {
         #[max_length = 32]
         merchant_connector_id -> Nullable<Varchar>,
         charges -> Nullable<Jsonb>,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 
diff --git a/crates/diesel_models/src/schema_v2.rs b/crates/diesel_models/src/schema_v2.rs
index 818415a277b..013157d5e17 100644
--- a/crates/diesel_models/src/schema_v2.rs
+++ b/crates/diesel_models/src/schema_v2.rs
@@ -385,6 +385,8 @@ diesel::table! {
         #[max_length = 32]
         merchant_connector_id -> Nullable<Varchar>,
         dispute_amount -> Int8,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 
@@ -810,6 +812,12 @@ diesel::table! {
         #[max_length = 64]
         client_version -> Nullable<Varchar>,
         customer_acceptance -> Nullable<Jsonb>,
+        #[max_length = 64]
+        profile_id -> Varchar,
+        #[max_length = 32]
+        organization_id -> Varchar,
+        #[max_length = 32]
+        card_network -> Nullable<Varchar>,
     }
 }
 
@@ -885,6 +893,8 @@ diesel::table! {
         merchant_order_reference_id -> Nullable<Varchar>,
         shipping_details -> Nullable<Bytea>,
         is_payment_processor_token_flow -> Nullable<Bool>,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 
@@ -1131,6 +1141,8 @@ diesel::table! {
         #[max_length = 32]
         merchant_connector_id -> Nullable<Varchar>,
         charges -> Nullable<Jsonb>,
+        #[max_length = 32]
+        organization_id -> Varchar,
     }
 }
 
diff --git a/crates/diesel_models/src/user/sample_data.rs b/crates/diesel_models/src/user/sample_data.rs
index dbdae51c45a..89f8af3d94e 100644
--- a/crates/diesel_models/src/user/sample_data.rs
+++ b/crates/diesel_models/src/user/sample_data.rs
@@ -80,6 +80,8 @@ pub struct PaymentAttemptBatchNew {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub customer_acceptance: Option<common_utils::pii::SecretSerdeValue>,
+    pub profile_id: common_utils::id_type::ProfileId,
+    pub organization_id: common_utils::id_type::OrganizationId,
 }
 
 #[allow(dead_code)]
@@ -117,6 +119,15 @@ impl PaymentAttemptBatchNew {
             connector_metadata: self.connector_metadata,
             payment_experience: self.payment_experience,
             payment_method_type: self.payment_method_type,
+            card_network: self
+                .payment_method_data
+                .as_ref()
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card"))
+                .and_then(|v| v.as_object())
+                .and_then(|v| v.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
             payment_method_data: self.payment_method_data,
             business_sub_label: self.business_sub_label,
             straight_through_algorithm: self.straight_through_algorithm,
@@ -144,6 +155,8 @@ impl PaymentAttemptBatchNew {
             client_source: self.client_source,
             client_version: self.client_version,
             customer_acceptance: self.customer_acceptance,
+            profile_id: self.profile_id,
+            organization_id: self.organization_id,
         }
     }
 }
diff --git a/crates/hyperswitch_domain_models/src/payments.rs b/crates/hyperswitch_domain_models/src/payments.rs
index 11a2c3c3164..3873dfcd291 100644
--- a/crates/hyperswitch_domain_models/src/payments.rs
+++ b/crates/hyperswitch_domain_models/src/payments.rs
@@ -67,4 +67,5 @@ pub struct PaymentIntent {
     pub merchant_order_reference_id: Option<String>,
     pub shipping_details: Option<Encryptable<Secret<serde_json::Value>>>,
     pub is_payment_processor_token_flow: Option<bool>,
+    pub organization_id: id_type::OrganizationId,
 }
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
index 12880172d3d..370b37d5a13 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_attempt.rs
@@ -181,6 +181,8 @@ pub struct PaymentAttempt {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub profile_id: id_type::ProfileId,
+    pub organization_id: id_type::OrganizationId,
 }
 
 impl PaymentAttempt {
@@ -206,7 +208,7 @@ pub struct PaymentListFilters {
     pub authentication_type: Vec<storage_enums::AuthenticationType>,
 }
 
-#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct PaymentAttemptNew {
     pub payment_id: String,
     pub merchant_id: id_type::MerchantId,
@@ -271,6 +273,8 @@ pub struct PaymentAttemptNew {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub customer_acceptance: Option<pii::SecretSerdeValue>,
+    pub profile_id: id_type::ProfileId,
+    pub organization_id: id_type::OrganizationId,
 }
 
 impl PaymentAttemptNew {
@@ -543,6 +547,7 @@ impl behaviour::Conversion for PaymentIntent {
             merchant_order_reference_id: self.merchant_order_reference_id,
             shipping_details: self.shipping_details.map(Encryption::from),
             is_payment_processor_token_flow: self.is_payment_processor_token_flow,
+            organization_id: self.organization_id,
         })
     }
     async fn convert_back(
@@ -625,6 +630,7 @@ impl behaviour::Conversion for PaymentIntent {
                     .async_lift(inner_decrypt)
                     .await?,
                 is_payment_processor_token_flow: storage_model.is_payment_processor_token_flow,
+                organization_id: storage_model.organization_id,
             })
         }
         .await
@@ -683,6 +689,7 @@ impl behaviour::Conversion for PaymentIntent {
             merchant_order_reference_id: self.merchant_order_reference_id,
             shipping_details: self.shipping_details.map(Encryption::from),
             is_payment_processor_token_flow: self.is_payment_processor_token_flow,
+            organization_id: self.organization_id,
         })
     }
 }
@@ -743,6 +750,7 @@ impl behaviour::Conversion for PaymentIntent {
             merchant_order_reference_id: self.merchant_order_reference_id,
             shipping_details: self.shipping_details.map(Encryption::from),
             is_payment_processor_token_flow: self.is_payment_processor_token_flow,
+            organization_id: self.organization_id,
         })
     }
 
@@ -826,6 +834,7 @@ impl behaviour::Conversion for PaymentIntent {
                     .async_lift(inner_decrypt)
                     .await?,
                 is_payment_processor_token_flow: storage_model.is_payment_processor_token_flow,
+                organization_id: storage_model.organization_id,
             })
         }
         .await
@@ -884,6 +893,7 @@ impl behaviour::Conversion for PaymentIntent {
             merchant_order_reference_id: self.merchant_order_reference_id,
             shipping_details: self.shipping_details.map(Encryption::from),
             is_payment_processor_token_flow: self.is_payment_processor_token_flow,
+            organization_id: self.organization_id,
         })
     }
 }
diff --git a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
index dcb06733cc8..f857e241e11 100644
--- a/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
+++ b/crates/hyperswitch_domain_models/src/payments/payment_intent.rs
@@ -151,6 +151,7 @@ pub struct PaymentIntentNew {
     pub billing_details: Option<Encryptable<Secret<serde_json::Value>>>,
     pub shipping_details: Option<Encryptable<Secret<serde_json::Value>>>,
     pub is_payment_processor_token_flow: Option<bool>,
+    pub organization_id: id_type::OrganizationId,
 }
 
 #[derive(Debug, Clone, Serialize)]
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 181e2232d90..ae84d4b0ade 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -2960,7 +2960,7 @@ mod tests {
             feature_metadata: None,
             attempt_count: 1,
             payment_link_id: None,
-            profile_id: None,
+            profile_id: Some(common_utils::generate_profile_id_of_default_length()),
             merchant_decision: None,
             payment_confirm_source: None,
             surcharge_applicable: None,
@@ -2982,6 +2982,7 @@ mod tests {
             merchant_order_reference_id: None,
             shipping_details: None,
             is_payment_processor_token_flow: None,
+            organization_id: id_type::OrganizationId::default(),
         };
         let req_cs = Some("1".to_string());
         assert!(authenticate_client_secret(req_cs.as_ref(), &payment_intent).is_ok());
@@ -3026,7 +3027,7 @@ mod tests {
             feature_metadata: None,
             attempt_count: 1,
             payment_link_id: None,
-            profile_id: None,
+            profile_id: Some(common_utils::generate_profile_id_of_default_length()),
             merchant_decision: None,
             payment_confirm_source: None,
             surcharge_applicable: None,
@@ -3047,6 +3048,7 @@ mod tests {
             merchant_order_reference_id: None,
             shipping_details: None,
             is_payment_processor_token_flow: None,
+            organization_id: id_type::OrganizationId::default(),
         };
         let req_cs = Some("1".to_string());
         assert!(authenticate_client_secret(req_cs.as_ref(), &payment_intent,).is_err())
@@ -3088,7 +3090,7 @@ mod tests {
             feature_metadata: None,
             attempt_count: 1,
             payment_link_id: None,
-            profile_id: None,
+            profile_id: Some(common_utils::generate_profile_id_of_default_length()),
             merchant_decision: None,
             payment_confirm_source: None,
             surcharge_applicable: None,
@@ -3110,6 +3112,7 @@ mod tests {
             merchant_order_reference_id: None,
             shipping_details: None,
             is_payment_processor_token_flow: None,
+            organization_id: id_type::OrganizationId::default(),
         };
         let req_cs = Some("1".to_string());
         assert!(authenticate_client_secret(req_cs.as_ref(), &payment_intent).is_err())
@@ -3645,6 +3648,8 @@ impl AttemptType {
             client_source: old_payment_attempt.client_source,
             client_version: old_payment_attempt.client_version,
             customer_acceptance: old_payment_attempt.customer_acceptance,
+            organization_id: old_payment_attempt.organization_id,
+            profile_id: old_payment_attempt.profile_id,
         }
     }
 
diff --git a/crates/router/src/core/payments/operations/payment_create.rs b/crates/router/src/core/payments/operations/payment_create.rs
index 0ba859f3e17..8ab3a957b06 100644
--- a/crates/router/src/core/payments/operations/payment_create.rs
+++ b/crates/router/src/core/payments/operations/payment_create.rs
@@ -308,6 +308,7 @@ impl<F: Send + Clone> GetTracker<F, PaymentData<F>, api::PaymentsRequest> for Pa
         let (payment_attempt_new, additional_payment_data) = Self::make_payment_attempt(
             &payment_id,
             merchant_id,
+            &merchant_account.organization_id,
             money,
             payment_method,
             payment_method_type,
@@ -874,6 +875,7 @@ impl PaymentCreate {
     pub async fn make_payment_attempt(
         payment_id: &str,
         merchant_id: &common_utils::id_type::MerchantId,
+        organization_id: &common_utils::id_type::OrganizationId,
         money: (api::Amount, enums::Currency),
         payment_method: Option<enums::PaymentMethod>,
         payment_method_type: Option<enums::PaymentMethodType>,
@@ -1065,6 +1067,8 @@ impl PaymentCreate {
                     .change_context(errors::ApiErrorResponse::InternalServerError)
                     .attach_printable("Failed to serialize customer_acceptance")?
                     .map(Secret::new),
+                organization_id: organization_id.clone(),
+                profile_id,
             },
             additional_pm_data,
         ))
@@ -1246,6 +1250,7 @@ impl PaymentCreate {
             merchant_order_reference_id: request.merchant_order_reference_id.clone(),
             shipping_details,
             is_payment_processor_token_flow,
+            organization_id: merchant_account.organization_id.clone(),
         })
     }
 
diff --git a/crates/router/src/core/payments/retry.rs b/crates/router/src/core/payments/retry.rs
index e67f30a2f79..d655fc42a10 100644
--- a/crates/router/src/core/payments/retry.rs
+++ b/crates/router/src/core/payments/retry.rs
@@ -506,7 +506,6 @@ pub fn make_new_payment_attempt(
         amount: old_payment_attempt.amount,
         currency: old_payment_attempt.currency,
         save_to_locker: old_payment_attempt.save_to_locker,
-
         offer_amount: old_payment_attempt.offer_amount,
         surcharge_amount: old_payment_attempt.surcharge_amount,
         tax_amount: old_payment_attempt.tax_amount,
@@ -521,7 +520,6 @@ pub fn make_new_payment_attempt(
         } else {
             old_payment_attempt.authentication_type
         },
-
         amount_to_capture: old_payment_attempt.amount_to_capture,
         mandate_id: old_payment_attempt.mandate_id,
         browser_info: old_payment_attempt.browser_info,
@@ -531,7 +529,37 @@ pub fn make_new_payment_attempt(
         created_at,
         modified_at,
         last_synced,
-        ..storage::PaymentAttemptNew::default()
+        net_amount: Default::default(),
+        error_message: Default::default(),
+        cancellation_reason: Default::default(),
+        error_code: Default::default(),
+        connector_metadata: Default::default(),
+        payment_experience: Default::default(),
+        payment_method_data: Default::default(),
+        business_sub_label: Default::default(),
+        straight_through_algorithm: Default::default(),
+        preprocessing_step_id: Default::default(),
+        mandate_details: Default::default(),
+        error_reason: Default::default(),
+        connector_response_reference_id: Default::default(),
+        multiple_capture_count: Default::default(),
+        amount_capturable: Default::default(),
+        updated_by: Default::default(),
+        authentication_data: Default::default(),
+        encoded_data: Default::default(),
+        merchant_connector_id: Default::default(),
+        unified_code: Default::default(),
+        unified_message: Default::default(),
+        external_three_ds_authentication_attempted: Default::default(),
+        authentication_connector: Default::default(),
+        authentication_id: Default::default(),
+        mandate_data: Default::default(),
+        payment_method_billing_address_id: Default::default(),
+        fingerprint_id: Default::default(),
+        charge_id: Default::default(),
+        customer_acceptance: Default::default(),
+        profile_id: old_payment_attempt.profile_id,
+        organization_id: old_payment_attempt.organization_id,
     }
 }
 
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index 89450347404..867a2ccff54 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -798,30 +798,34 @@ pub async fn validate_and_create_refund(
         .clone()
         .ok_or(errors::ApiErrorResponse::InternalServerError)
         .attach_printable("No connector populated in payment attempt")?;
-
-    let refund_create_req = storage::RefundNew::default()
-        .set_refund_id(refund_id.to_string())
-        .set_internal_reference_id(utils::generate_id(consts::ID_LENGTH, "refid"))
-        .set_external_reference_id(Some(refund_id.clone()))
-        .set_payment_id(req.payment_id)
-        .set_merchant_id(merchant_account.get_id().clone())
-        .set_connector_transaction_id(connecter_transaction_id.to_string())
-        .set_connector(connector)
-        .set_refund_type(req.refund_type.unwrap_or_default().foreign_into())
-        .set_total_amount(payment_attempt.amount)
-        .set_refund_amount(refund_amount)
-        .set_currency(currency)
-        .set_created_at(common_utils::date_time::now())
-        .set_modified_at(common_utils::date_time::now())
-        .set_refund_status(enums::RefundStatus::Pending)
-        .set_metadata(req.metadata)
-        .set_description(req.reason.clone())
-        .set_attempt_id(payment_attempt.attempt_id.clone())
-        .set_refund_reason(req.reason)
-        .set_profile_id(payment_intent.profile_id.clone())
-        .set_merchant_connector_id(payment_attempt.merchant_connector_id.clone())
-        .set_charges(req.charges)
-        .to_owned();
+    let refund_create_req = storage::RefundNew {
+        refund_id: refund_id.to_string(),
+        internal_reference_id: utils::generate_id(consts::ID_LENGTH, "refid"),
+        external_reference_id: Some(refund_id.clone()),
+        payment_id: req.payment_id,
+        merchant_id: merchant_account.get_id().clone(),
+        connector_transaction_id: connecter_transaction_id.to_string(),
+        connector,
+        refund_type: req.refund_type.unwrap_or_default().foreign_into(),
+        total_amount: payment_attempt.amount,
+        refund_amount,
+        currency,
+        created_at: common_utils::date_time::now(),
+        modified_at: common_utils::date_time::now(),
+        refund_status: enums::RefundStatus::Pending,
+        metadata: req.metadata,
+        description: req.reason.clone(),
+        attempt_id: payment_attempt.attempt_id.clone(),
+        refund_reason: req.reason,
+        profile_id: payment_intent.profile_id.clone(),
+        merchant_connector_id: payment_attempt.merchant_connector_id.clone(),
+        charges: req.charges,
+        connector_refund_id: None,
+        sent_to_gateway: Default::default(),
+        refund_arn: None,
+        updated_by: Default::default(),
+        organization_id: merchant_account.organization_id.clone(),
+    };
 
     let refund = match db
         .insert_refund(refund_create_req, merchant_account.storage_scheme)
diff --git a/crates/router/src/core/user/sample_data.rs b/crates/router/src/core/user/sample_data.rs
index f5faf68435f..ce7143338f4 100644
--- a/crates/router/src/core/user/sample_data.rs
+++ b/crates/router/src/core/user/sample_data.rs
@@ -19,7 +19,13 @@ pub async fn generate_sample_data_for_user(
     req: SampleDataRequest,
     _req_state: ReqState,
 ) -> SampleDataApiResponse<()> {
-    let sample_data = generate_sample_data(&state, req, &user_from_token.merchant_id).await?;
+    let sample_data = generate_sample_data(
+        &state,
+        req,
+        &user_from_token.merchant_id,
+        &user_from_token.org_id,
+    )
+    .await?;
 
     let key_store = state
         .store
diff --git a/crates/router/src/core/webhooks/incoming.rs b/crates/router/src/core/webhooks/incoming.rs
index 0a0a35c23cc..79d6d5d6b82 100644
--- a/crates/router/src/core/webhooks/incoming.rs
+++ b/crates/router/src/core/webhooks/incoming.rs
@@ -902,6 +902,7 @@ async fn get_or_update_dispute_object(
     option_dispute: Option<diesel_models::dispute::Dispute>,
     dispute_details: api::disputes::DisputePayload,
     merchant_id: &common_utils::id_type::MerchantId,
+    organization_id: &common_utils::id_type::OrganizationId,
     payment_attempt: &hyperswitch_domain_models::payments::payment_attempt::PaymentAttempt,
     event_type: webhooks::IncomingWebhookEvent,
     business_profile: &domain::BusinessProfile,
@@ -935,6 +936,7 @@ async fn get_or_update_dispute_object(
                 evidence: None,
                 merchant_connector_id: payment_attempt.merchant_connector_id.clone(),
                 dispute_amount: dispute_details.amount.parse::<i64>().unwrap_or(0),
+                organization_id: organization_id.clone(),
             };
             state
                 .store
@@ -1377,6 +1379,7 @@ async fn disputes_incoming_webhook_flow(
             option_dispute,
             dispute_details,
             merchant_account.get_id(),
+            &merchant_account.organization_id,
             &payment_attempt,
             event_type,
             &business_profile,
diff --git a/crates/router/src/db/dispute.rs b/crates/router/src/db/dispute.rs
index 9d6093bfe3c..686de8c2233 100644
--- a/crates/router/src/db/dispute.rs
+++ b/crates/router/src/db/dispute.rs
@@ -170,6 +170,7 @@ impl DisputeInterface for MockDb {
             evidence,
             merchant_connector_id: dispute.merchant_connector_id,
             dispute_amount: dispute.dispute_amount,
+            organization_id: dispute.organization_id,
         };
 
         locked_disputes.push(new_dispute.clone());
@@ -404,9 +405,10 @@ mod tests {
                 connector_updated_at: Some(datetime!(2019-01-03 0:00)),
                 connector: "connector".into(),
                 evidence: Some(Secret::from(Value::String("evidence".into()))),
-                profile_id: None,
+                profile_id: Some(common_utils::generate_profile_id_of_default_length()),
                 merchant_connector_id: None,
                 dispute_amount: 1040,
+                organization_id: common_utils::id_type::OrganizationId::default(),
             }
         }
 
diff --git a/crates/router/src/db/refund.rs b/crates/router/src/db/refund.rs
index 1fa7e28cb81..408c6495609 100644
--- a/crates/router/src/db/refund.rs
+++ b/crates/router/src/db/refund.rs
@@ -398,6 +398,7 @@ mod storage {
                         updated_by: new.updated_by.clone(),
                         merchant_connector_id: new.merchant_connector_id.clone(),
                         charges: new.charges.clone(),
+                        organization_id: new.organization_id.clone(),
                     };
 
                     let field = format!(
@@ -859,6 +860,7 @@ impl RefundInterface for MockDb {
             updated_by: new.updated_by,
             merchant_connector_id: new.merchant_connector_id,
             charges: new.charges,
+            organization_id: new.organization_id,
         };
         refunds.push(refund.clone());
         Ok(refund)
diff --git a/crates/router/src/services/kafka/dispute.rs b/crates/router/src/services/kafka/dispute.rs
index 62a4310a640..9374c42cbe4 100644
--- a/crates/router/src/services/kafka/dispute.rs
+++ b/crates/router/src/services/kafka/dispute.rs
@@ -1,3 +1,4 @@
+use common_utils::id_type;
 use diesel_models::enums as storage_enums;
 use masking::Secret;
 use time::OffsetDateTime;
@@ -13,7 +14,7 @@ pub struct KafkaDispute<'a> {
     pub dispute_status: &'a storage_enums::DisputeStatus,
     pub payment_id: &'a String,
     pub attempt_id: &'a String,
-    pub merchant_id: &'a common_utils::id_type::MerchantId,
+    pub merchant_id: &'a id_type::MerchantId,
     pub connector_status: &'a String,
     pub connector_dispute_id: &'a String,
     pub connector_reason: Option<&'a String>,
@@ -30,8 +31,9 @@ pub struct KafkaDispute<'a> {
     pub modified_at: OffsetDateTime,
     pub connector: &'a String,
     pub evidence: &'a Secret<serde_json::Value>,
-    pub profile_id: Option<&'a common_utils::id_type::ProfileId>,
-    pub merchant_connector_id: Option<&'a common_utils::id_type::MerchantConnectorAccountId>,
+    pub profile_id: Option<&'a id_type::ProfileId>,
+    pub merchant_connector_id: Option<&'a id_type::MerchantConnectorAccountId>,
+    pub organization_id: &'a id_type::OrganizationId,
 }
 
 impl<'a> KafkaDispute<'a> {
@@ -58,6 +60,7 @@ impl<'a> KafkaDispute<'a> {
             evidence: &dispute.evidence,
             profile_id: dispute.profile_id.as_ref(),
             merchant_connector_id: dispute.merchant_connector_id.as_ref(),
+            organization_id: &dispute.organization_id,
         }
     }
 }
diff --git a/crates/router/src/services/kafka/dispute_event.rs b/crates/router/src/services/kafka/dispute_event.rs
index 9b73cc2e510..ac72af0cd95 100644
--- a/crates/router/src/services/kafka/dispute_event.rs
+++ b/crates/router/src/services/kafka/dispute_event.rs
@@ -33,6 +33,7 @@ pub struct KafkaDisputeEvent<'a> {
     pub evidence: &'a Secret<serde_json::Value>,
     pub profile_id: Option<&'a common_utils::id_type::ProfileId>,
     pub merchant_connector_id: Option<&'a common_utils::id_type::MerchantConnectorAccountId>,
+    pub organization_id: &'a common_utils::id_type::OrganizationId,
 }
 
 impl<'a> KafkaDisputeEvent<'a> {
@@ -59,6 +60,7 @@ impl<'a> KafkaDisputeEvent<'a> {
             evidence: &dispute.evidence,
             profile_id: dispute.profile_id.as_ref(),
             merchant_connector_id: dispute.merchant_connector_id.as_ref(),
+            organization_id: &dispute.organization_id,
         }
     }
 }
diff --git a/crates/router/src/services/kafka/payment_attempt.rs b/crates/router/src/services/kafka/payment_attempt.rs
index 08ec50f06b7..b92728196d4 100644
--- a/crates/router/src/services/kafka/payment_attempt.rs
+++ b/crates/router/src/services/kafka/payment_attempt.rs
@@ -1,5 +1,5 @@
 // use diesel_models::enums::MandateDetails;
-use common_utils::types::MinorUnit;
+use common_utils::{id_type, types::MinorUnit};
 use diesel_models::enums as storage_enums;
 use hyperswitch_domain_models::{
     mandates::MandateDetails, payments::payment_attempt::PaymentAttempt,
@@ -9,7 +9,7 @@ use time::OffsetDateTime;
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaPaymentAttempt<'a> {
     pub payment_id: &'a String,
-    pub merchant_id: &'a common_utils::id_type::MerchantId,
+    pub merchant_id: &'a id_type::MerchantId,
     pub attempt_id: &'a String,
     pub status: storage_enums::AttemptStatus,
     pub amount: MinorUnit,
@@ -47,13 +47,16 @@ pub struct KafkaPaymentAttempt<'a> {
     pub error_reason: Option<&'a String>,
     pub multiple_capture_count: Option<i16>,
     pub amount_capturable: MinorUnit,
-    pub merchant_connector_id: Option<&'a common_utils::id_type::MerchantConnectorAccountId>,
+    pub merchant_connector_id: Option<&'a id_type::MerchantConnectorAccountId>,
     pub net_amount: MinorUnit,
     pub unified_code: Option<&'a String>,
     pub unified_message: Option<&'a String>,
     pub mandate_data: Option<&'a MandateDetails>,
     pub client_source: Option<&'a String>,
     pub client_version: Option<&'a String>,
+    pub profile_id: &'a id_type::ProfileId,
+    pub organization_id: &'a id_type::OrganizationId,
+    pub card_network: Option<String>,
 }
 
 impl<'a> KafkaPaymentAttempt<'a> {
@@ -100,6 +103,17 @@ impl<'a> KafkaPaymentAttempt<'a> {
             mandate_data: attempt.mandate_data.as_ref(),
             client_source: attempt.client_source.as_ref(),
             client_version: attempt.client_version.as_ref(),
+            profile_id: &attempt.profile_id,
+            organization_id: &attempt.organization_id,
+            card_network: attempt
+                .payment_method_data
+                .as_ref()
+                .and_then(|data| data.as_object())
+                .and_then(|pm| pm.get("card"))
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
         }
     }
 }
diff --git a/crates/router/src/services/kafka/payment_attempt_event.rs b/crates/router/src/services/kafka/payment_attempt_event.rs
index c6337a3b5f0..e9f8fbdfc5c 100644
--- a/crates/router/src/services/kafka/payment_attempt_event.rs
+++ b/crates/router/src/services/kafka/payment_attempt_event.rs
@@ -1,5 +1,5 @@
 // use diesel_models::enums::MandateDetails;
-use common_utils::types::MinorUnit;
+use common_utils::{id_type, types::MinorUnit};
 use diesel_models::enums as storage_enums;
 use hyperswitch_domain_models::{
     mandates::MandateDetails, payments::payment_attempt::PaymentAttempt,
@@ -10,7 +10,7 @@ use time::OffsetDateTime;
 #[derive(serde::Serialize, Debug)]
 pub struct KafkaPaymentAttemptEvent<'a> {
     pub payment_id: &'a String,
-    pub merchant_id: &'a common_utils::id_type::MerchantId,
+    pub merchant_id: &'a id_type::MerchantId,
     pub attempt_id: &'a String,
     pub status: storage_enums::AttemptStatus,
     pub amount: MinorUnit,
@@ -48,13 +48,16 @@ pub struct KafkaPaymentAttemptEvent<'a> {
     pub error_reason: Option<&'a String>,
     pub multiple_capture_count: Option<i16>,
     pub amount_capturable: MinorUnit,
-    pub merchant_connector_id: Option<&'a common_utils::id_type::MerchantConnectorAccountId>,
+    pub merchant_connector_id: Option<&'a id_type::MerchantConnectorAccountId>,
     pub net_amount: MinorUnit,
     pub unified_code: Option<&'a String>,
     pub unified_message: Option<&'a String>,
     pub mandate_data: Option<&'a MandateDetails>,
     pub client_source: Option<&'a String>,
     pub client_version: Option<&'a String>,
+    pub profile_id: &'a id_type::ProfileId,
+    pub organization_id: &'a id_type::OrganizationId,
+    pub card_network: Option<String>,
 }
 
 impl<'a> KafkaPaymentAttemptEvent<'a> {
@@ -101,6 +104,17 @@ impl<'a> KafkaPaymentAttemptEvent<'a> {
             mandate_data: attempt.mandate_data.as_ref(),
             client_source: attempt.client_source.as_ref(),
             client_version: attempt.client_version.as_ref(),
+            profile_id: &attempt.profile_id,
+            organization_id: &attempt.organization_id,
+            card_network: attempt
+                .payment_method_data
+                .as_ref()
+                .and_then(|data| data.as_object())
+                .and_then(|pm| pm.get("card"))
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
         }
     }
 }
diff --git a/crates/router/src/services/kafka/payment_intent.rs b/crates/router/src/services/kafka/payment_intent.rs
index 29aa5632837..31f424b2a28 100644
--- a/crates/router/src/services/kafka/payment_intent.rs
+++ b/crates/router/src/services/kafka/payment_intent.rs
@@ -40,6 +40,7 @@ pub struct KafkaPaymentIntent<'a> {
     pub customer_email: Option<HashedString<pii::EmailStrategy>>,
     pub feature_metadata: Option<&'a Value>,
     pub merchant_order_reference_id: Option<&'a String>,
+    pub organization_id: &'a id_type::OrganizationId,
 }
 
 impl<'a> KafkaPaymentIntent<'a> {
@@ -82,6 +83,7 @@ impl<'a> KafkaPaymentIntent<'a> {
                 .map(|email| HashedString::from(Secret::new(email.to_string()))),
             feature_metadata: intent.feature_metadata.as_ref(),
             merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
+            organization_id: &intent.organization_id,
         }
     }
 }
diff --git a/crates/router/src/services/kafka/payment_intent_event.rs b/crates/router/src/services/kafka/payment_intent_event.rs
index 50d2c48a4bd..f92f3fa9326 100644
--- a/crates/router/src/services/kafka/payment_intent_event.rs
+++ b/crates/router/src/services/kafka/payment_intent_event.rs
@@ -41,6 +41,7 @@ pub struct KafkaPaymentIntentEvent<'a> {
     pub customer_email: Option<HashedString<pii::EmailStrategy>>,
     pub feature_metadata: Option<&'a Value>,
     pub merchant_order_reference_id: Option<&'a String>,
+    pub organization_id: &'a id_type::OrganizationId,
 }
 
 impl<'a> KafkaPaymentIntentEvent<'a> {
@@ -83,6 +84,7 @@ impl<'a> KafkaPaymentIntentEvent<'a> {
                 .map(|email| HashedString::from(Secret::new(email.to_string()))),
             feature_metadata: intent.feature_metadata.as_ref(),
             merchant_order_reference_id: intent.merchant_order_reference_id.as_ref(),
+            organization_id: &intent.organization_id,
         }
     }
 }
diff --git a/crates/router/src/services/kafka/refund.rs b/crates/router/src/services/kafka/refund.rs
index 1aaf5d2ccb3..09537812e57 100644
--- a/crates/router/src/services/kafka/refund.rs
+++ b/crates/router/src/services/kafka/refund.rs
@@ -1,4 +1,4 @@
-use common_utils::types::MinorUnit;
+use common_utils::{id_type, types::MinorUnit};
 use diesel_models::{enums as storage_enums, refund::Refund};
 use time::OffsetDateTime;
 
@@ -7,7 +7,7 @@ pub struct KafkaRefund<'a> {
     pub internal_reference_id: &'a String,
     pub refund_id: &'a String, //merchant_reference id
     pub payment_id: &'a String,
-    pub merchant_id: &'a common_utils::id_type::MerchantId,
+    pub merchant_id: &'a id_type::MerchantId,
     pub connector_transaction_id: &'a String,
     pub connector: &'a String,
     pub connector_refund_id: Option<&'a String>,
@@ -28,6 +28,8 @@ pub struct KafkaRefund<'a> {
     pub attempt_id: &'a String,
     pub refund_reason: Option<&'a String>,
     pub refund_error_code: Option<&'a String>,
+    pub profile_id: Option<&'a id_type::ProfileId>,
+    pub organization_id: &'a id_type::OrganizationId,
 }
 
 impl<'a> KafkaRefund<'a> {
@@ -55,6 +57,8 @@ impl<'a> KafkaRefund<'a> {
             attempt_id: &refund.attempt_id,
             refund_reason: refund.refund_reason.as_ref(),
             refund_error_code: refund.refund_error_code.as_ref(),
+            profile_id: refund.profile_id.as_ref(),
+            organization_id: &refund.organization_id,
         }
     }
 }
diff --git a/crates/router/src/services/kafka/refund_event.rs b/crates/router/src/services/kafka/refund_event.rs
index 9480220104c..764146f808f 100644
--- a/crates/router/src/services/kafka/refund_event.rs
+++ b/crates/router/src/services/kafka/refund_event.rs
@@ -1,4 +1,4 @@
-use common_utils::types::MinorUnit;
+use common_utils::{id_type, types::MinorUnit};
 use diesel_models::{enums as storage_enums, refund::Refund};
 use time::OffsetDateTime;
 
@@ -8,7 +8,7 @@ pub struct KafkaRefundEvent<'a> {
     pub internal_reference_id: &'a String,
     pub refund_id: &'a String, //merchant_reference id
     pub payment_id: &'a String,
-    pub merchant_id: &'a common_utils::id_type::MerchantId,
+    pub merchant_id: &'a id_type::MerchantId,
     pub connector_transaction_id: &'a String,
     pub connector: &'a String,
     pub connector_refund_id: Option<&'a String>,
@@ -29,6 +29,8 @@ pub struct KafkaRefundEvent<'a> {
     pub attempt_id: &'a String,
     pub refund_reason: Option<&'a String>,
     pub refund_error_code: Option<&'a String>,
+    pub profile_id: Option<&'a id_type::ProfileId>,
+    pub organization_id: &'a id_type::OrganizationId,
 }
 
 impl<'a> KafkaRefundEvent<'a> {
@@ -56,6 +58,8 @@ impl<'a> KafkaRefundEvent<'a> {
             attempt_id: &refund.attempt_id,
             refund_reason: refund.refund_reason.as_ref(),
             refund_error_code: refund.refund_error_code.as_ref(),
+            profile_id: refund.profile_id.as_ref(),
+            organization_id: &refund.organization_id,
         }
     }
 }
diff --git a/crates/router/src/types/storage/payment_attempt.rs b/crates/router/src/types/storage/payment_attempt.rs
index 669427d88c7..6775f9a81a7 100644
--- a/crates/router/src/types/storage/payment_attempt.rs
+++ b/crates/router/src/types/storage/payment_attempt.rs
@@ -129,7 +129,60 @@ mod tests {
             connector: Some(connector),
             created_at: current_time.into(),
             modified_at: current_time.into(),
-            ..PaymentAttemptNew::default()
+            merchant_id: Default::default(),
+            attempt_id: Default::default(),
+            status: Default::default(),
+            amount: Default::default(),
+            net_amount: Default::default(),
+            currency: Default::default(),
+            save_to_locker: Default::default(),
+            error_message: Default::default(),
+            offer_amount: Default::default(),
+            surcharge_amount: Default::default(),
+            tax_amount: Default::default(),
+            payment_method_id: Default::default(),
+            payment_method: Default::default(),
+            capture_method: Default::default(),
+            capture_on: Default::default(),
+            confirm: Default::default(),
+            authentication_type: Default::default(),
+            last_synced: Default::default(),
+            cancellation_reason: Default::default(),
+            amount_to_capture: Default::default(),
+            mandate_id: Default::default(),
+            browser_info: Default::default(),
+            payment_token: Default::default(),
+            error_code: Default::default(),
+            connector_metadata: Default::default(),
+            payment_experience: Default::default(),
+            payment_method_type: Default::default(),
+            payment_method_data: Default::default(),
+            business_sub_label: Default::default(),
+            straight_through_algorithm: Default::default(),
+            preprocessing_step_id: Default::default(),
+            mandate_details: Default::default(),
+            error_reason: Default::default(),
+            connector_response_reference_id: Default::default(),
+            multiple_capture_count: Default::default(),
+            amount_capturable: Default::default(),
+            updated_by: Default::default(),
+            authentication_data: Default::default(),
+            encoded_data: Default::default(),
+            merchant_connector_id: Default::default(),
+            unified_code: Default::default(),
+            unified_message: Default::default(),
+            external_three_ds_authentication_attempted: Default::default(),
+            authentication_connector: Default::default(),
+            authentication_id: Default::default(),
+            mandate_data: Default::default(),
+            payment_method_billing_address_id: Default::default(),
+            fingerprint_id: Default::default(),
+            charge_id: Default::default(),
+            client_source: Default::default(),
+            client_version: Default::default(),
+            customer_acceptance: Default::default(),
+            profile_id: common_utils::generate_profile_id_of_default_length(),
+            organization_id: Default::default(),
         };
 
         let store = state
@@ -163,7 +216,58 @@ mod tests {
             created_at: current_time.into(),
             modified_at: current_time.into(),
             attempt_id: attempt_id.clone(),
-            ..PaymentAttemptNew::default()
+            status: Default::default(),
+            amount: Default::default(),
+            net_amount: Default::default(),
+            currency: Default::default(),
+            save_to_locker: Default::default(),
+            error_message: Default::default(),
+            offer_amount: Default::default(),
+            surcharge_amount: Default::default(),
+            tax_amount: Default::default(),
+            payment_method_id: Default::default(),
+            payment_method: Default::default(),
+            capture_method: Default::default(),
+            capture_on: Default::default(),
+            confirm: Default::default(),
+            authentication_type: Default::default(),
+            last_synced: Default::default(),
+            cancellation_reason: Default::default(),
+            amount_to_capture: Default::default(),
+            mandate_id: Default::default(),
+            browser_info: Default::default(),
+            payment_token: Default::default(),
+            error_code: Default::default(),
+            connector_metadata: Default::default(),
+            payment_experience: Default::default(),
+            payment_method_type: Default::default(),
+            payment_method_data: Default::default(),
+            business_sub_label: Default::default(),
+            straight_through_algorithm: Default::default(),
+            preprocessing_step_id: Default::default(),
+            mandate_details: Default::default(),
+            error_reason: Default::default(),
+            connector_response_reference_id: Default::default(),
+            multiple_capture_count: Default::default(),
+            amount_capturable: Default::default(),
+            updated_by: Default::default(),
+            authentication_data: Default::default(),
+            encoded_data: Default::default(),
+            merchant_connector_id: Default::default(),
+            unified_code: Default::default(),
+            unified_message: Default::default(),
+            external_three_ds_authentication_attempted: Default::default(),
+            authentication_connector: Default::default(),
+            authentication_id: Default::default(),
+            mandate_data: Default::default(),
+            payment_method_billing_address_id: Default::default(),
+            fingerprint_id: Default::default(),
+            charge_id: Default::default(),
+            client_source: Default::default(),
+            client_version: Default::default(),
+            customer_acceptance: Default::default(),
+            profile_id: common_utils::generate_profile_id_of_default_length(),
+            organization_id: Default::default(),
         };
         let store = state
             .stores
@@ -207,10 +311,59 @@ mod tests {
             connector: Some(connector),
             created_at: current_time.into(),
             modified_at: current_time.into(),
-            // Adding a mandate_id
             mandate_id: Some("man_121212".to_string()),
             attempt_id: uuid.clone(),
-            ..PaymentAttemptNew::default()
+            status: Default::default(),
+            amount: Default::default(),
+            net_amount: Default::default(),
+            currency: Default::default(),
+            save_to_locker: Default::default(),
+            error_message: Default::default(),
+            offer_amount: Default::default(),
+            surcharge_amount: Default::default(),
+            tax_amount: Default::default(),
+            payment_method_id: Default::default(),
+            payment_method: Default::default(),
+            capture_method: Default::default(),
+            capture_on: Default::default(),
+            confirm: Default::default(),
+            authentication_type: Default::default(),
+            last_synced: Default::default(),
+            cancellation_reason: Default::default(),
+            amount_to_capture: Default::default(),
+            browser_info: Default::default(),
+            payment_token: Default::default(),
+            error_code: Default::default(),
+            connector_metadata: Default::default(),
+            payment_experience: Default::default(),
+            payment_method_type: Default::default(),
+            payment_method_data: Default::default(),
+            business_sub_label: Default::default(),
+            straight_through_algorithm: Default::default(),
+            preprocessing_step_id: Default::default(),
+            mandate_details: Default::default(),
+            error_reason: Default::default(),
+            connector_response_reference_id: Default::default(),
+            multiple_capture_count: Default::default(),
+            amount_capturable: Default::default(),
+            updated_by: Default::default(),
+            authentication_data: Default::default(),
+            encoded_data: Default::default(),
+            merchant_connector_id: Default::default(),
+            unified_code: Default::default(),
+            unified_message: Default::default(),
+            external_three_ds_authentication_attempted: Default::default(),
+            authentication_connector: Default::default(),
+            authentication_id: Default::default(),
+            mandate_data: Default::default(),
+            payment_method_billing_address_id: Default::default(),
+            fingerprint_id: Default::default(),
+            charge_id: Default::default(),
+            client_source: Default::default(),
+            client_version: Default::default(),
+            customer_acceptance: Default::default(),
+            profile_id: common_utils::generate_profile_id_of_default_length(),
+            organization_id: Default::default(),
         };
         let store = state
             .stores
diff --git a/crates/router/src/utils/user/sample_data.rs b/crates/router/src/utils/user/sample_data.rs
index ac9272c86d7..94890a41ad7 100644
--- a/crates/router/src/utils/user/sample_data.rs
+++ b/crates/router/src/utils/user/sample_data.rs
@@ -20,6 +20,7 @@ pub async fn generate_sample_data(
     state: &SessionState,
     req: SampleDataRequest,
     merchant_id: &id_type::MerchantId,
+    org_id: &id_type::OrganizationId,
 ) -> SampleDataResult<Vec<(PaymentIntent, PaymentAttemptBatchNew, Option<RefundNew>)>> {
     let sample_data_size: usize = req.record.unwrap_or(100);
     let key_manager_state = &state.into();
@@ -252,6 +253,7 @@ pub async fn generate_sample_data(
             merchant_order_reference_id: Default::default(),
             shipping_details: None,
             is_payment_processor_token_flow: None,
+            organization_id: org_id.clone(),
         };
         let payment_attempt = PaymentAttemptBatchNew {
             attempt_id: attempt_id.clone(),
@@ -329,6 +331,8 @@ pub async fn generate_sample_data(
             client_source: None,
             client_version: None,
             customer_acceptance: None,
+            profile_id: profile_id.clone(),
+            organization_id: org_id.clone(),
         };
 
         let refund = if refunds_count < number_of_refunds && !is_failed_payment {
@@ -364,6 +368,7 @@ pub async fn generate_sample_data(
                 updated_by: merchant_from_db.storage_scheme.to_string(),
                 merchant_connector_id: payment_attempt.merchant_connector_id.clone(),
                 charges: None,
+                organization_id: org_id.clone(),
             })
         } else {
             None
diff --git a/crates/storage_impl/src/mock_db/payment_attempt.rs b/crates/storage_impl/src/mock_db/payment_attempt.rs
index 37e01cbbb5c..67840636de8 100644
--- a/crates/storage_impl/src/mock_db/payment_attempt.rs
+++ b/crates/storage_impl/src/mock_db/payment_attempt.rs
@@ -158,6 +158,8 @@ impl PaymentAttemptInterface for MockDb {
             client_source: payment_attempt.client_source,
             client_version: payment_attempt.client_version,
             customer_acceptance: payment_attempt.customer_acceptance,
+            organization_id: payment_attempt.organization_id,
+            profile_id: payment_attempt.profile_id,
         };
         payment_attempts.push(payment_attempt.clone());
         Ok(payment_attempt)
diff --git a/crates/storage_impl/src/payments/payment_attempt.rs b/crates/storage_impl/src/payments/payment_attempt.rs
index 382abacfa78..63b10cf01bd 100644
--- a/crates/storage_impl/src/payments/payment_attempt.rs
+++ b/crates/storage_impl/src/payments/payment_attempt.rs
@@ -419,6 +419,8 @@ impl<T: DatabaseStore> PaymentAttemptInterface for KVRouterStore<T> {
                     client_source: payment_attempt.client_source.clone(),
                     client_version: payment_attempt.client_version.clone(),
                     customer_acceptance: payment_attempt.customer_acceptance.clone(),
+                    organization_id: payment_attempt.organization_id.clone(),
+                    profile_id: payment_attempt.profile_id.clone(),
                 };
 
                 let field = format!("pa_{}", created_attempt.attempt_id);
@@ -1189,6 +1191,15 @@ impl DataModelExt for PaymentAttempt {
             connector_metadata: self.connector_metadata,
             payment_experience: self.payment_experience,
             payment_method_type: self.payment_method_type,
+            card_network: self
+                .payment_method_data
+                .as_ref()
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card"))
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
             payment_method_data: self.payment_method_data,
             business_sub_label: self.business_sub_label,
             straight_through_algorithm: self.straight_through_algorithm,
@@ -1215,6 +1226,8 @@ impl DataModelExt for PaymentAttempt {
             client_source: self.client_source,
             client_version: self.client_version,
             customer_acceptance: self.customer_acceptance,
+            organization_id: self.organization_id,
+            profile_id: self.profile_id,
         }
     }
 
@@ -1282,6 +1295,8 @@ impl DataModelExt for PaymentAttempt {
             client_source: storage_model.client_source,
             client_version: storage_model.client_version,
             customer_acceptance: storage_model.customer_acceptance,
+            organization_id: storage_model.organization_id,
+            profile_id: storage_model.profile_id,
         }
     }
 }
@@ -1330,6 +1345,15 @@ impl DataModelExt for PaymentAttempt {
             connector_metadata: self.connector_metadata,
             payment_experience: self.payment_experience,
             payment_method_type: self.payment_method_type,
+            card_network: self
+                .payment_method_data
+                .as_ref()
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card"))
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
             payment_method_data: self.payment_method_data,
             business_sub_label: self.business_sub_label,
             straight_through_algorithm: self.straight_through_algorithm,
@@ -1356,6 +1380,8 @@ impl DataModelExt for PaymentAttempt {
             client_source: self.client_source,
             client_version: self.client_version,
             customer_acceptance: self.customer_acceptance,
+            organization_id: self.organization_id,
+            profile_id: self.profile_id,
         }
     }
 
@@ -1423,6 +1449,8 @@ impl DataModelExt for PaymentAttempt {
             client_source: storage_model.client_source,
             client_version: storage_model.client_version,
             customer_acceptance: storage_model.customer_acceptance,
+            organization_id: storage_model.organization_id,
+            profile_id: storage_model.profile_id,
         }
     }
 }
@@ -1471,6 +1499,15 @@ impl DataModelExt for PaymentAttemptNew {
             connector_metadata: self.connector_metadata,
             payment_experience: self.payment_experience,
             payment_method_type: self.payment_method_type,
+            card_network: self
+                .payment_method_data
+                .as_ref()
+                .and_then(|data| data.as_object())
+                .and_then(|card| card.get("card"))
+                .and_then(|value| value.as_object())
+                .and_then(|map| map.get("card_network"))
+                .and_then(|network| network.as_str())
+                .map(|network| network.to_string()),
             payment_method_data: self.payment_method_data,
             business_sub_label: self.business_sub_label,
             straight_through_algorithm: self.straight_through_algorithm,
@@ -1497,6 +1534,8 @@ impl DataModelExt for PaymentAttemptNew {
             client_source: self.client_source,
             client_version: self.client_version,
             customer_acceptance: self.customer_acceptance,
+            organization_id: self.organization_id,
+            profile_id: self.profile_id,
         }
     }
 
@@ -1563,6 +1602,8 @@ impl DataModelExt for PaymentAttemptNew {
             client_source: storage_model.client_source,
             client_version: storage_model.client_version,
             customer_acceptance: storage_model.customer_acceptance,
+            organization_id: storage_model.organization_id,
+            profile_id: storage_model.profile_id,
         }
     }
 }
diff --git a/migrations/2024-08-20-112035_add-profile-id-to-txn-tables/down.sql b/migrations/2024-08-20-112035_add-profile-id-to-txn-tables/down.sql
new file mode 100644
index 00000000000..66707bebb4e
--- /dev/null
+++ b/migrations/2024-08-20-112035_add-profile-id-to-txn-tables/down.sql
@@ -0,0 +1,20 @@
+-- This file should undo anything in `up.sql`
+-- Remove profile_id from payment_attempt table
+ALTER TABLE payment_attempt
+DROP COLUMN IF EXISTS profile_id;
+
+-- Remove organization_id from payment_attempt table
+ALTER TABLE payment_attempt
+DROP COLUMN IF EXISTS organization_id;
+
+-- Remove organization_id from payment_intent table
+ALTER TABLE payment_intent
+DROP COLUMN IF EXISTS organization_id;
+
+-- Remove organization_id from refund table
+ALTER TABLE refund
+DROP COLUMN IF EXISTS organization_id;
+
+-- Remove organization_id from dispute table
+ALTER TABLE dispute
+DROP COLUMN IF EXISTS organization_id;
\ No newline at end of file
diff --git a/migrations/2024-08-20-112035_add-profile-id-to-txn-tables/up.sql b/migrations/2024-08-20-112035_add-profile-id-to-txn-tables/up.sql
new file mode 100644
index 00000000000..74f55308fc6
--- /dev/null
+++ b/migrations/2024-08-20-112035_add-profile-id-to-txn-tables/up.sql
@@ -0,0 +1,43 @@
+-- Your SQL goes here
+ALTER TABLE payment_attempt
+ADD COLUMN IF NOT EXISTS profile_id VARCHAR(64) NOT NULL DEFAULT 'default_profile';
+
+-- Add organization_id to payment_attempt table
+ALTER TABLE payment_attempt
+ADD COLUMN IF NOT EXISTS organization_id VARCHAR(32) NOT NULL DEFAULT 'default_org';
+
+-- Add organization_id to payment_intent table
+ALTER TABLE payment_intent
+ADD COLUMN IF NOT EXISTS organization_id VARCHAR(32) NOT NULL DEFAULT 'default_org';
+
+-- Add organization_id to refund table
+ALTER TABLE refund
+ADD COLUMN IF NOT EXISTS organization_id VARCHAR(32) NOT NULL DEFAULT 'default_org';
+
+-- Add organization_id to dispute table
+ALTER TABLE dispute
+ADD COLUMN IF NOT EXISTS organization_id VARCHAR(32) NOT NULL DEFAULT 'default_org';
+
+-- This doesn't work on V2
+-- The below backfill step has to be run after the code deployment
+-- UPDATE payment_attempt pa
+-- SET organization_id = ma.organization_id
+-- FROM merchant_account ma
+-- WHERE pa.merchant_id = ma.merchant_id;
+
+-- UPDATE payment_intent pi
+-- SET organization_id = ma.organization_id
+-- FROM merchant_account ma
+-- WHERE pi.merchant_id = ma.merchant_id;
+
+-- UPDATE refund r
+-- SET organization_id = ma.organization_id
+-- FROM merchant_account ma
+-- WHERE r.merchant_id = ma.merchant_id;
+
+-- UPDATE payment_attempt pa
+-- SET profile_id = pi.profile_id
+-- FROM payment_intent pi
+-- WHERE pa.payment_id = pi.payment_id
+--   AND pa.merchant_id = pi.merchant_id
+--   AND pi.profile_id IS NOT NULL;
diff --git a/migrations/2024-08-26-043046_add-card-network-field-for-pa/down.sql b/migrations/2024-08-26-043046_add-card-network-field-for-pa/down.sql
new file mode 100644
index 00000000000..8dc04221730
--- /dev/null
+++ b/migrations/2024-08-26-043046_add-card-network-field-for-pa/down.sql
@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+
+ALTER TABLE payment_attempt DROP COLUMN card_network;
\ No newline at end of file
diff --git a/migrations/2024-08-26-043046_add-card-network-field-for-pa/up.sql b/migrations/2024-08-26-043046_add-card-network-field-for-pa/up.sql
new file mode 100644
index 00000000000..43f29b56d72
--- /dev/null
+++ b/migrations/2024-08-26-043046_add-card-network-field-for-pa/up.sql
@@ -0,0 +1,5 @@
+-- Your SQL goes here
+
+ALTER TABLE payment_attempt ADD COLUMN card_network VARCHAR(32);
+UPDATE payment_attempt
+SET card_network = (payment_method_data -> 'card' -> 'card_network')::VARCHAR(32);
\ No newline at end of file