Issue:
Bug: [Analytics] - Add error_reason field in clickhouse column





Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [x] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Adding new fields in Kafka struct for analytics usecase -->
This PR adds more fields which we would want to send to Kafka for various usecases of search and analytics from attempts table in PSQL

Fields which have been added are
```
payment_method_data,
error_reason,
multiple_capture_count,
amount_capturable,
merchant_connector_id,
net_amount,
unified_code,
unified_message,
mandate_data,
```

### 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).
-->


## 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)?
-->
Curl 
```curl
curl --location 'http://localhost:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: key' \
--data-raw '{
  "amount": 60,
  "currency": "USD",
  "confirm": true,
  "capture_method": "automatic",
  "capture_on": "2022-09-10T10:11:12Z",
  "amount_to_capture": 60,
  "customer_id": "SHIVANSH",
  "email": "guest@example.com",
  "name": "John Doe",
  "phone": "999999999",
  "phone_country_code": "+1",
  "description": "Its my first payment request",
  "authentication_type": "no_three_ds",
  "return_url": "https://google.com",
  "payment_method": "card",
  "payment_method_type": "credit",
  "payment_method_data": {
    "card": {
      "card_number": "4242424242424242",
      "card_exp_month": "10",
      "card_exp_year": "25",
      "card_holder_name": "joseph Doe",
      "card_cvc": "123"
    }
  },
  "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"
    }
  },
  "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"
    }
  },
  "statement_descriptor_name": "joseph",
  "statement_descriptor_suffix": "JS",
  "metadata": {
    "udf1": "value1",
    "new_customer": "true",
    "login_date": "2019-09-10T10:11:12Z"
  }
}'
```

![Screenshot 2024-02-26 at 12 38 45 PM](https://github.com/juspay/hyperswitch/assets/104988143/0b67e04a-3290-4943-99bf-9c2741191862)

Tested it locally , by creating payments and observing what events are being generated for that payment which is being pushed to Kafka

Can test the changes in loki though once deployed 
As soon as we deploy this
The fields which we have
We’ll get those fields under `topic="hyperswitch-payment-attempt-events"`

## 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
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable


Testing:

Curl 
```curl
curl --location 'http://localhost:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: key' \
--data-raw '{
  "amount": 60,
  "currency": "USD",
  "confirm": true,
  "capture_method": "automatic",
  "capture_on": "2022-09-10T10:11:12Z",
  "amount_to_capture": 60,
  "customer_id": "SHIVANSH",
  "email": "guest@example.com",
  "name": "John Doe",
  "phone": "999999999",
  "phone_country_code": "+1",
  "description": "Its my first payment request",
  "authentication_type": "no_three_ds",
  "return_url": "https://google.com",
  "payment_method": "card",
  "payment_method_type": "credit",
  "payment_method_data": {
    "card": {
      "card_number": "4242424242424242",
      "card_exp_month": "10",
      "card_exp_year": "25",
      "card_holder_name": "joseph Doe",
      "card_cvc": "123"
    }
  },
  "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"
    }
  },
  "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"
    }
  },
  "statement_descriptor_name": "joseph",
  "statement_descriptor_suffix": "JS",
  "metadata": {
    "udf1": "value1",
    "new_customer": "true",
    "login_date": "2019-09-10T10:11:12Z"
  }
}'
```

![Screenshot 2024-02-26 at 12 38 45 PM](https://github.com/juspay/hyperswitch/assets/104988143/0b67e04a-3290-4943-99bf-9c2741191862)

Tested it locally , by creating payments and observing what events are being generated for that payment which is being pushed to Kafka

Can test the changes in loki though once deployed 
As soon as we deploy this
The fields which we have
We’ll get those fields under `topic="hyperswitch-payment-attempt-events"`

Patch:
diff --git a/crates/analytics/docs/clickhouse/cluster_setup/scripts/dispute_analytics.sql b/crates/analytics/docs/clickhouse/cluster_setup/scripts/dispute_analytics.sql
deleted file mode 100644
index 92a748ff489..00000000000
--- a/crates/analytics/docs/clickhouse/cluster_setup/scripts/dispute_analytics.sql
+++ /dev/null
@@ -1,142 +0,0 @@
-CREATE TABLE hyperswitch.dispute_queue on cluster '{cluster}' (
-    `dispute_id` String,
-    `amount` String,
-    `currency` String,
-    `dispute_stage` LowCardinality(String),
-    `dispute_status` LowCardinality(String),
-    `payment_id` String,
-    `attempt_id` String,
-    `merchant_id` String,
-    `connector_status` String,
-    `connector_dispute_id` String,
-    `connector_reason` Nullable(String),
-    `connector_reason_code` Nullable(String),
-    `challenge_required_by` Nullable(DateTime) CODEC(T64, LZ4),
-    `connector_created_at` Nullable(DateTime) CODEC(T64, LZ4),
-    `connector_updated_at` Nullable(DateTime) CODEC(T64, LZ4),
-    `created_at` DateTime CODEC(T64, LZ4),
-    `modified_at` DateTime CODEC(T64, LZ4),
-    `connector` LowCardinality(String),
-    `evidence` Nullable(String),
-    `profile_id` Nullable(String),
-    `merchant_connector_id` Nullable(String),
-    `sign_flag` Int8
-) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
-kafka_topic_list = 'hyperswitch-dispute-events',
-kafka_group_name = 'hyper-c1',
-kafka_format = 'JSONEachRow',
-kafka_handle_error_mode = 'stream';
-
-
-CREATE MATERIALIZED VIEW hyperswitch.dispute_mv on cluster '{cluster}' TO hyperswitch.dispute (
-    `dispute_id` String,
-    `amount` String,
-    `currency` String,
-    `dispute_stage` LowCardinality(String),
-    `dispute_status` LowCardinality(String),
-    `payment_id` String,
-    `attempt_id` String,
-    `merchant_id` String,
-    `connector_status` String,
-    `connector_dispute_id` String,
-    `connector_reason` Nullable(String),
-    `connector_reason_code` Nullable(String),
-    `challenge_required_by` Nullable(DateTime64(3)),
-    `connector_created_at` Nullable(DateTime64(3)),
-    `connector_updated_at` Nullable(DateTime64(3)),
-    `created_at` DateTime64(3),
-    `modified_at` DateTime64(3),
-    `connector` LowCardinality(String),
-    `evidence` Nullable(String),
-    `profile_id` Nullable(String),
-    `merchant_connector_id` Nullable(String),
-    `inserted_at` DateTime64(3),
-    `sign_flag` Int8
-) AS
-SELECT
-    dispute_id,
-    amount,
-    currency,
-    dispute_stage,
-    dispute_status,
-    payment_id,
-    attempt_id,
-    merchant_id,
-    connector_status,
-    connector_dispute_id,
-    connector_reason,
-    connector_reason_code,
-    challenge_required_by,
-    connector_created_at,
-    connector_updated_at,
-    created_at,
-    modified_at,
-    connector,
-    evidence,
-    profile_id,
-    merchant_connector_id,
-    now() as inserted_at,
-    sign_flag
-FROM
-    hyperswitch.dispute_queue
-WHERE length(_error) = 0;
-
-
-CREATE TABLE hyperswitch.dispute_clustered on cluster '{cluster}' (
-    `dispute_id` String,
-    `amount` String,
-    `currency` String,
-    `dispute_stage` LowCardinality(String),
-    `dispute_status` LowCardinality(String),
-    `payment_id` String,
-    `attempt_id` String,
-    `merchant_id` String,
-    `connector_status` String,
-    `connector_dispute_id` String,
-    `connector_reason` Nullable(String),
-    `connector_reason_code` Nullable(String),
-    `challenge_required_by` Nullable(DateTime) CODEC(T64, LZ4),
-    `connector_created_at` Nullable(DateTime) CODEC(T64, LZ4),
-    `connector_updated_at` Nullable(DateTime) CODEC(T64, LZ4),
-    `created_at` DateTime DEFAULT now() CODEC(T64, LZ4),
-    `modified_at` DateTime DEFAULT now() CODEC(T64, LZ4),
-    `connector` LowCardinality(String),
-    `evidence` String DEFAULT '{}' CODEC(T64, LZ4),
-    `profile_id` Nullable(String),
-    `merchant_connector_id` Nullable(String),
-    `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
-    `sign_flag` Int8
-    INDEX connectorIndex connector TYPE bloom_filter GRANULARITY 1,
-    INDEX disputeStatusIndex dispute_status TYPE bloom_filter GRANULARITY 1,
-    INDEX disputeStageIndex dispute_stage TYPE bloom_filter GRANULARITY 1
-) ENGINE = ReplicatedCollapsingMergeTree(
-    '/clickhouse/{installation}/{cluster}/tables/{shard}/hyperswitch/dispute_clustered',
-    '{replica}',
-    dispute_status
-)
-PARTITION BY toStartOfDay(created_at)
-ORDER BY
-    (created_at, merchant_id, dispute_id)
-TTL created_at + toIntervalMonth(6);
-
-
-CREATE MATERIALIZED VIEW hyperswitch.dispute_parse_errors on cluster '{cluster}'
-(
-    `topic` String,
-    `partition` Int64,
-    `offset` Int64,
-    `raw` String,
-    `error` String
-)
-ENGINE = MergeTree
-ORDER BY (topic, partition, offset)
-SETTINGS index_granularity = 8192 AS
-SELECT
-    _topic AS topic,
-    _partition AS partition,
-    _offset AS offset,
-    _raw_message AS raw,
-    _error AS error
-FROM hyperswitch.dispute_queue
-WHERE length(_error) > 0
-;
\ No newline at end of file
diff --git a/crates/analytics/docs/clickhouse/scripts/api_events.sql b/crates/analytics/docs/clickhouse/scripts/api_events.sql
index 49a6472eaa4..e466fc56cb6 100644
--- a/crates/analytics/docs/clickhouse/scripts/api_events.sql
+++ b/crates/analytics/docs/clickhouse/scripts/api_events.sql
@@ -58,7 +58,7 @@ CREATE TABLE api_events_dist (
     `hs_latency` Nullable(UInt128),
     `http_method` LowCardinality(String),
     `url_path` String,
-    `dispute_id` Nullable(String)
+    `dispute_id` Nullable(String),
     INDEX flowIndex flow_type TYPE bloom_filter GRANULARITY 1,
     INDEX apiIndex api_flow TYPE bloom_filter GRANULARITY 1,
     INDEX statusIndex status_code TYPE bloom_filter GRANULARITY 1
diff --git a/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql b/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql
index 276e311e57a..8b7715044c1 100644
--- a/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql
+++ b/crates/analytics/docs/clickhouse/scripts/payment_attempts.sql
@@ -29,6 +29,15 @@ CREATE TABLE payment_attempts_queue (
     `created_at` DateTime CODEC(T64, LZ4),
     `last_synced` Nullable(DateTime) CODEC(T64, LZ4),
     `modified_at` DateTime CODEC(T64, LZ4),
+    `payment_method_data` Nullable(String),
+    `error_reason` Nullable(String),
+    `multiple_capture_count` Nullable(Int16),
+    `amount_capturable` Nullable(UInt64) ,
+    `merchant_connector_id`  Nullable(String),
+    `net_amount` Nullable(UInt64) ,
+    `unified_code`  Nullable(String),
+    `unified_message`  Nullable(String),
+    `mandate_data`  Nullable(String),
     `sign_flag` Int8
 ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka0:29092',
 kafka_topic_list = 'hyperswitch-payment-attempt-events',
@@ -67,6 +76,15 @@ CREATE TABLE payment_attempt_dist (
     `created_at` DateTime DEFAULT now() CODEC(T64, LZ4),
     `last_synced` Nullable(DateTime) CODEC(T64, LZ4),
     `modified_at` DateTime DEFAULT now() CODEC(T64, LZ4),
+    `payment_method_data` Nullable(String),
+    `error_reason` Nullable(String),
+    `multiple_capture_count` Nullable(Int16),
+    `amount_capturable` Nullable(UInt64) ,
+    `merchant_connector_id`  Nullable(String),
+    `net_amount` Nullable(UInt64) ,
+    `unified_code`  Nullable(String),
+    `unified_message`  Nullable(String),
+    `mandate_data`  Nullable(String),
     `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
     `sign_flag` Int8,
     INDEX connectorIndex connector TYPE bloom_filter GRANULARITY 1,
@@ -115,6 +133,15 @@ CREATE MATERIALIZED VIEW kafka_parse_pa TO payment_attempt_dist (
     `capture_on` Nullable(DateTime64(3)),
     `last_synced` Nullable(DateTime64(3)),
     `modified_at` DateTime64(3),
+    `payment_method_data` Nullable(String),
+    `error_reason` Nullable(String),
+    `multiple_capture_count` Nullable(Int16),
+    `amount_capturable` Nullable(UInt64) ,
+    `merchant_connector_id`  Nullable(String),
+    `net_amount` Nullable(UInt64) ,
+    `unified_code`  Nullable(String),
+    `unified_message`  Nullable(String),
+    `mandate_data`  Nullable(String),
     `inserted_at` DateTime64(3),
     `sign_flag` Int8
 ) AS
@@ -149,6 +176,15 @@ SELECT
     capture_on,
     last_synced,
     modified_at,
+    payment_method_data,
+    error_reason,
+    multiple_capture_count,
+    amount_capturable,
+    merchant_connector_id,
+    net_amount,
+    unified_code,
+    unified_message,
+    mandate_data,
     now() as inserted_at,
     sign_flag
 FROM
diff --git a/crates/router/src/services/kafka/payment_attempt.rs b/crates/router/src/services/kafka/payment_attempt.rs
index ea0721f418e..98f88fb69f5 100644
--- a/crates/router/src/services/kafka/payment_attempt.rs
+++ b/crates/router/src/services/kafka/payment_attempt.rs
@@ -1,4 +1,5 @@
-use data_models::payments::payment_attempt::PaymentAttempt;
+// use diesel_models::enums::MandateDetails;
+use data_models::{mandates::MandateDetails, payments::payment_attempt::PaymentAttempt};
 use diesel_models::enums as storage_enums;
 use time::OffsetDateTime;
 
@@ -39,6 +40,15 @@ pub struct KafkaPaymentAttempt<'a> {
     // TODO: These types should implement copy ideally
     pub payment_experience: Option<&'a storage_enums::PaymentExperience>,
     pub payment_method_type: Option<&'a storage_enums::PaymentMethodType>,
+    pub payment_method_data: Option<String>,
+    pub error_reason: Option<&'a String>,
+    pub multiple_capture_count: Option<i16>,
+    pub amount_capturable: i64,
+    pub merchant_connector_id: Option<&'a String>,
+    pub net_amount: i64,
+    pub unified_code: Option<&'a String>,
+    pub unified_message: Option<&'a String>,
+    pub mandate_data: Option<&'a MandateDetails>,
 }
 
 impl<'a> KafkaPaymentAttempt<'a> {
@@ -74,6 +84,15 @@ impl<'a> KafkaPaymentAttempt<'a> {
             connector_metadata: attempt.connector_metadata.as_ref().map(|v| v.to_string()),
             payment_experience: attempt.payment_experience.as_ref(),
             payment_method_type: attempt.payment_method_type.as_ref(),
+            payment_method_data: attempt.payment_method_data.as_ref().map(|v| v.to_string()),
+            error_reason: attempt.error_reason.as_ref(),
+            multiple_capture_count: attempt.multiple_capture_count,
+            amount_capturable: attempt.amount_capturable,
+            merchant_connector_id: attempt.merchant_connector_id.as_ref(),
+            net_amount: attempt.net_amount,
+            unified_code: attempt.unified_code.as_ref(),
+            unified_message: attempt.unified_message.as_ref(),
+            mandate_data: attempt.mandate_data.as_ref(),
         }
     }
 }