Bug: feat(analytics): Add metrics, filters and APIs for Analytics v2 Dashboard - Payments Page



Add metrics, filters and APIs for Analytics v2 Dashboard - Payments page

The following filters should be added:

Payment Attempts:

- Merchant_id
- Card last 4
- Error Reason
- Card Issuer

Payment Intents:

- connector
- authentication_type
- payment_method
- payment_method_type
- card_network
- merchant id
- card_last_4
- error_reason
- card_issuer

New API endpoints for Payments Lifecycle (Sankey) on different levels should be created
diff --git a/crates/analytics/src/clickhouse.rs b/crates/analytics/src/clickhouse.rs
index 37a011c9a5a..546b57f99af 100644
--- a/crates/analytics/src/clickhouse.rs
+++ b/crates/analytics/src/clickhouse.rs
@@ -130,9 +130,12 @@ impl AnalyticsDataSource for ClickhouseClient {
     fn get_table_engine(table: AnalyticsCollection) -> TableEngine {
         match table {
             AnalyticsCollection::Payment
+            | AnalyticsCollection::PaymentSessionized
             | AnalyticsCollection::Refund
+            | AnalyticsCollection::RefundSessionized
             | AnalyticsCollection::FraudCheck
             | AnalyticsCollection::PaymentIntent
+            | AnalyticsCollection::PaymentIntentSessionized
             | AnalyticsCollection::Dispute => {
                 TableEngine::CollapsingMergeTree { sign: "sign_flag" }
             }
@@ -423,13 +426,16 @@ impl ToSql<ClickhouseClient> for AnalyticsCollection {
     fn to_sql(&self, _table_engine: &TableEngine) -> error_stack::Result<String, ParsingError> {
         match self {
             Self::Payment => Ok("payment_attempts".to_string()),
+            Self::PaymentSessionized => Ok("sessionizer_payment_attempts".to_string()),
             Self::Refund => Ok("refunds".to_string()),
+            Self::RefundSessionized => Ok("sessionizer_refunds".to_string()),
             Self::FraudCheck => Ok("fraud_check".to_string()),
             Self::SdkEvents => Ok("sdk_events_audit".to_string()),
             Self::SdkEventsAnalytics => Ok("sdk_events".to_string()),
             Self::ApiEvents => Ok("api_events_audit".to_string()),
             Self::ApiEventsAnalytics => Ok("api_events".to_string()),
             Self::PaymentIntent => Ok("payment_intents".to_string()),
+            Self::PaymentIntentSessionized => Ok("sessionizer_payment_intents".to_string()),
             Self::ConnectorEvents => Ok("connector_events_audit".to_string()),
             Self::OutgoingWebhookEvent => Ok("outgoing_webhook_events_audit".to_string()),
             Self::Dispute => Ok("dispute".to_string()),
diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs
index d5cb3c718df..224cd82ccd3 100644
--- a/crates/analytics/src/lib.rs
+++ b/crates/analytics/src/lib.rs
@@ -996,6 +996,7 @@ pub enum AnalyticsFlow {
     GetSearchResults,
     GetDisputeFilters,
     GetDisputeMetrics,
+    GetSankey,
 }
 
 impl FlowMetric for AnalyticsFlow {}
diff --git a/crates/analytics/src/payment_intents.rs b/crates/analytics/src/payment_intents.rs
index 449dd94788c..809be1b20a1 100644
--- a/crates/analytics/src/payment_intents.rs
+++ b/crates/analytics/src/payment_intents.rs
@@ -2,6 +2,7 @@ pub mod accumulator;
 mod core;
 pub mod filters;
 pub mod metrics;
+pub mod sankey;
 pub mod types;
 pub use accumulator::{PaymentIntentMetricAccumulator, PaymentIntentMetricsAccumulator};
 
@@ -10,4 +11,4 @@ pub trait PaymentIntentAnalytics:
 {
 }
 
-pub use self::core::{get_filters, get_metrics};
+pub use self::core::{get_filters, get_metrics, get_sankey};
diff --git a/crates/analytics/src/payment_intents/accumulator.rs b/crates/analytics/src/payment_intents/accumulator.rs
index 8fd98a1e73c..cbb8335cea0 100644
--- a/crates/analytics/src/payment_intents/accumulator.rs
+++ b/crates/analytics/src/payment_intents/accumulator.rs
@@ -1,5 +1,6 @@
 use api_models::analytics::payment_intents::PaymentIntentMetricsBucketValue;
 use bigdecimal::ToPrimitive;
+use diesel_models::enums as storage_enums;
 
 use super::metrics::PaymentIntentMetricRow;
 
@@ -7,8 +8,11 @@ use super::metrics::PaymentIntentMetricRow;
 pub struct PaymentIntentMetricsAccumulator {
     pub successful_smart_retries: CountAccumulator,
     pub total_smart_retries: CountAccumulator,
-    pub smart_retried_amount: SumAccumulator,
+    pub smart_retried_amount: SmartRetriedAmountAccumulator,
     pub payment_intent_count: CountAccumulator,
+    pub payments_success_rate: PaymentsSuccessRateAccumulator,
+    pub payment_processed_amount: ProcessedAmountAccumulator,
+    pub payments_distribution: PaymentsDistributionAccumulator,
 }
 
 #[derive(Debug, Default)]
@@ -38,9 +42,31 @@ pub trait PaymentIntentMetricAccumulator {
 }
 
 #[derive(Debug, Default)]
-#[repr(transparent)]
-pub struct SumAccumulator {
-    pub total: Option<i64>,
+pub struct SmartRetriedAmountAccumulator {
+    pub amount: Option<i64>,
+    pub amount_without_retries: Option<i64>,
+}
+
+#[derive(Debug, Default)]
+pub struct PaymentsSuccessRateAccumulator {
+    pub success: u32,
+    pub success_without_retries: u32,
+    pub total: u32,
+}
+
+#[derive(Debug, Default)]
+pub struct ProcessedAmountAccumulator {
+    pub count_with_retries: Option<i64>,
+    pub total_with_retries: Option<i64>,
+    pub count_without_retries: Option<i64>,
+    pub total_without_retries: Option<i64>,
+}
+
+#[derive(Debug, Default)]
+pub struct PaymentsDistributionAccumulator {
+    pub success_without_retries: u32,
+    pub failed_without_retries: u32,
+    pub total: u32,
 }
 
 impl PaymentIntentMetricAccumulator for CountAccumulator {
@@ -59,32 +85,251 @@ impl PaymentIntentMetricAccumulator for CountAccumulator {
     }
 }
 
-impl PaymentIntentMetricAccumulator for SumAccumulator {
-    type MetricOutput = Option<u64>;
+impl PaymentIntentMetricAccumulator for SmartRetriedAmountAccumulator {
+    type MetricOutput = (Option<u64>, Option<u64>);
     #[inline]
     fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
-        self.total = match (
-            self.total,
+        self.amount = match (
+            self.amount,
             metrics.total.as_ref().and_then(ToPrimitive::to_i64),
         ) {
             (None, None) => None,
             (None, i @ Some(_)) | (i @ Some(_), None) => i,
             (Some(a), Some(b)) => Some(a + b),
+        };
+        if metrics.first_attempt.unwrap_or(0) == 1 {
+            self.amount_without_retries = match (
+                self.amount_without_retries,
+                metrics.total.as_ref().and_then(ToPrimitive::to_i64),
+            ) {
+                (None, None) => None,
+                (None, i @ Some(_)) | (i @ Some(_), None) => i,
+                (Some(a), Some(b)) => Some(a + b),
+            }
+        } else {
+            self.amount_without_retries = Some(0);
         }
     }
     #[inline]
     fn collect(self) -> Self::MetricOutput {
-        self.total.and_then(|i| u64::try_from(i).ok())
+        let with_retries = self.amount.and_then(|i| u64::try_from(i).ok()).or(Some(0));
+        let without_retries = self
+            .amount_without_retries
+            .and_then(|i| u64::try_from(i).ok())
+            .or(Some(0));
+        (with_retries, without_retries)
+    }
+}
+
+impl PaymentIntentMetricAccumulator for PaymentsSuccessRateAccumulator {
+    type MetricOutput = (
+        Option<u32>,
+        Option<u32>,
+        Option<u32>,
+        Option<f64>,
+        Option<f64>,
+    );
+
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
+        if let Some(ref status) = metrics.status {
+            if status.as_ref() == &storage_enums::IntentStatus::Succeeded {
+                if let Some(success) = metrics
+                    .count
+                    .and_then(|success| u32::try_from(success).ok())
+                {
+                    self.success += success;
+                    if metrics.first_attempt.unwrap_or(0) == 1 {
+                        self.success_without_retries += success;
+                    }
+                }
+            }
+            if status.as_ref() != &storage_enums::IntentStatus::RequiresCustomerAction
+                && status.as_ref() != &storage_enums::IntentStatus::RequiresPaymentMethod
+                && status.as_ref() != &storage_enums::IntentStatus::RequiresMerchantAction
+                && status.as_ref() != &storage_enums::IntentStatus::RequiresConfirmation
+            {
+                if let Some(total) = metrics.count.and_then(|total| u32::try_from(total).ok()) {
+                    self.total += total;
+                }
+            }
+        }
+    }
+
+    fn collect(self) -> Self::MetricOutput {
+        if self.total == 0 {
+            (None, None, None, None, None)
+        } else {
+            let success = Some(self.success);
+            let success_without_retries = Some(self.success_without_retries);
+            let total = Some(self.total);
+
+            let success_rate = match (success, total) {
+                (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                _ => None,
+            };
+
+            let success_without_retries_rate = match (success_without_retries, total) {
+                (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                _ => None,
+            };
+
+            (
+                success,
+                success_without_retries,
+                total,
+                success_rate,
+                success_without_retries_rate,
+            )
+        }
+    }
+}
+
+impl PaymentIntentMetricAccumulator for ProcessedAmountAccumulator {
+    type MetricOutput = (Option<u64>, Option<u64>, Option<u64>, Option<u64>);
+    #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
+        self.total_with_retries = match (
+            self.total_with_retries,
+            metrics.total.as_ref().and_then(ToPrimitive::to_i64),
+        ) {
+            (None, None) => None,
+            (None, i @ Some(_)) | (i @ Some(_), None) => i,
+            (Some(a), Some(b)) => Some(a + b),
+        };
+
+        self.count_with_retries = match (self.count_with_retries, metrics.count) {
+            (None, None) => None,
+            (None, i @ Some(_)) | (i @ Some(_), None) => i,
+            (Some(a), Some(b)) => Some(a + b),
+        };
+
+        if metrics.first_attempt.unwrap_or(0) == 1 {
+            self.total_without_retries = match (
+                self.total_without_retries,
+                metrics.total.as_ref().and_then(ToPrimitive::to_i64),
+            ) {
+                (None, None) => None,
+                (None, i @ Some(_)) | (i @ Some(_), None) => i,
+                (Some(a), Some(b)) => Some(a + b),
+            };
+
+            self.count_without_retries = match (self.count_without_retries, metrics.count) {
+                (None, None) => None,
+                (None, i @ Some(_)) | (i @ Some(_), None) => i,
+                (Some(a), Some(b)) => Some(a + b),
+            };
+        }
+    }
+    #[inline]
+    fn collect(self) -> Self::MetricOutput {
+        let total_with_retries = u64::try_from(self.total_with_retries.unwrap_or(0)).ok();
+        let count_with_retries = self.count_with_retries.and_then(|i| u64::try_from(i).ok());
+
+        let total_without_retries = u64::try_from(self.total_without_retries.unwrap_or(0)).ok();
+        let count_without_retries = self
+            .count_without_retries
+            .and_then(|i| u64::try_from(i).ok());
+
+        (
+            total_with_retries,
+            count_with_retries,
+            total_without_retries,
+            count_without_retries,
+        )
+    }
+}
+
+impl PaymentIntentMetricAccumulator for PaymentsDistributionAccumulator {
+    type MetricOutput = (Option<f64>, Option<f64>);
+
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
+        let first_attempt = metrics.first_attempt.unwrap_or(0);
+        if let Some(ref status) = metrics.status {
+            if status.as_ref() == &storage_enums::IntentStatus::Succeeded {
+                if let Some(success) = metrics
+                    .count
+                    .and_then(|success| u32::try_from(success).ok())
+                {
+                    if first_attempt == 1 {
+                        self.success_without_retries += success;
+                    }
+                }
+            }
+            if let Some(failed) = metrics.count.and_then(|failed| u32::try_from(failed).ok()) {
+                if first_attempt == 0
+                    || (first_attempt == 1
+                        && status.as_ref() == &storage_enums::IntentStatus::Failed)
+                {
+                    self.failed_without_retries += failed;
+                }
+            }
+
+            if let Some(total) = metrics.count.and_then(|total| u32::try_from(total).ok()) {
+                self.total += total;
+            }
+        }
+    }
+
+    fn collect(self) -> Self::MetricOutput {
+        if self.total == 0 {
+            (None, None)
+        } else {
+            let success_without_retries = Some(self.success_without_retries);
+            let failed_without_retries = Some(self.failed_without_retries);
+            let total = Some(self.total);
+
+            let success_rate_without_retries = match (success_without_retries, total) {
+                (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                _ => None,
+            };
+
+            let failed_rate_without_retries = match (failed_without_retries, total) {
+                (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                _ => None,
+            };
+            (success_rate_without_retries, failed_rate_without_retries)
+        }
     }
 }
 
 impl PaymentIntentMetricsAccumulator {
     pub fn collect(self) -> PaymentIntentMetricsBucketValue {
+        let (
+            successful_payments,
+            successful_payments_without_smart_retries,
+            total_payments,
+            payments_success_rate,
+            payments_success_rate_without_smart_retries,
+        ) = self.payments_success_rate.collect();
+        let (smart_retried_amount, smart_retried_amount_without_smart_retries) =
+            self.smart_retried_amount.collect();
+        let (
+            payment_processed_amount,
+            payment_processed_count,
+            payment_processed_amount_without_smart_retries,
+            payment_processed_count_without_smart_retries,
+        ) = self.payment_processed_amount.collect();
+        let (
+            payments_success_rate_distribution_without_smart_retries,
+            payments_failure_rate_distribution_without_smart_retries,
+        ) = self.payments_distribution.collect();
         PaymentIntentMetricsBucketValue {
             successful_smart_retries: self.successful_smart_retries.collect(),
             total_smart_retries: self.total_smart_retries.collect(),
-            smart_retried_amount: self.smart_retried_amount.collect(),
+            smart_retried_amount,
+            smart_retried_amount_without_smart_retries,
             payment_intent_count: self.payment_intent_count.collect(),
+            successful_payments,
+            successful_payments_without_smart_retries,
+            total_payments,
+            payments_success_rate,
+            payments_success_rate_without_smart_retries,
+            payment_processed_amount,
+            payment_processed_count,
+            payment_processed_amount_without_smart_retries,
+            payment_processed_count_without_smart_retries,
+            payments_success_rate_distribution_without_smart_retries,
+            payments_failure_rate_distribution_without_smart_retries,
         }
     }
 }
diff --git a/crates/analytics/src/payment_intents/core.rs b/crates/analytics/src/payment_intents/core.rs
index f8a5c48986a..e04c3b7bd9e 100644
--- a/crates/analytics/src/payment_intents/core.rs
+++ b/crates/analytics/src/payment_intents/core.rs
@@ -6,10 +6,12 @@ use api_models::analytics::{
         MetricsBucketResponse, PaymentIntentDimensions, PaymentIntentMetrics,
         PaymentIntentMetricsBucketIdentifier,
     },
-    AnalyticsMetadata, GetPaymentIntentFiltersRequest, GetPaymentIntentMetricRequest,
-    MetricsResponse, PaymentIntentFilterValue, PaymentIntentFiltersResponse,
+    GetPaymentIntentFiltersRequest, GetPaymentIntentMetricRequest, PaymentIntentFilterValue,
+    PaymentIntentFiltersResponse, PaymentIntentsAnalyticsMetadata, PaymentIntentsMetricsResponse,
+    SankeyResponse,
 };
-use common_utils::errors::CustomResult;
+use common_enums::IntentStatus;
+use common_utils::{errors::CustomResult, types::TimeRange};
 use error_stack::ResultExt;
 use router_env::{
     instrument, logger,
@@ -20,6 +22,7 @@ use router_env::{
 use super::{
     filters::{get_payment_intent_filter_for_dimension, PaymentIntentFilterRow},
     metrics::PaymentIntentMetricRow,
+    sankey::{get_sankey_data, SessionizerRefundStatus},
     PaymentIntentMetricsAccumulator,
 };
 use crate::{
@@ -41,12 +44,85 @@ pub enum TaskType {
     ),
 }
 
+#[instrument(skip_all)]
+pub async fn get_sankey(
+    pool: &AnalyticsProvider,
+    auth: &AuthInfo,
+    req: TimeRange,
+) -> AnalyticsResult<SankeyResponse> {
+    match pool {
+        AnalyticsProvider::Sqlx(_) => Err(AnalyticsError::NotImplemented(
+            "Sankey not implemented for sqlx",
+        ))?,
+        AnalyticsProvider::Clickhouse(ckh_pool)
+        | AnalyticsProvider::CombinedCkh(_, ckh_pool)
+        | AnalyticsProvider::CombinedSqlx(_, ckh_pool) => {
+            let sankey_rows = get_sankey_data(ckh_pool, auth, &req)
+                .await
+                .change_context(AnalyticsError::UnknownError)?;
+            let mut sankey_response = SankeyResponse::default();
+            for i in sankey_rows {
+                match (
+                    i.status.as_ref(),
+                    i.refunds_status.unwrap_or_default().as_ref(),
+                    i.attempt_count,
+                ) {
+                    (IntentStatus::Succeeded, SessionizerRefundStatus::FullRefunded, _) => {
+                        sankey_response.refunded += i.count
+                    }
+                    (IntentStatus::Succeeded, SessionizerRefundStatus::PartialRefunded, _) => {
+                        sankey_response.partial_refunded += i.count
+                    }
+                    (
+                        IntentStatus::Succeeded
+                        | IntentStatus::PartiallyCaptured
+                        | IntentStatus::PartiallyCapturedAndCapturable
+                        | IntentStatus::RequiresCapture,
+                        SessionizerRefundStatus::NotRefunded,
+                        1,
+                    ) => sankey_response.normal_success += i.count,
+                    (
+                        IntentStatus::Succeeded
+                        | IntentStatus::PartiallyCaptured
+                        | IntentStatus::PartiallyCapturedAndCapturable
+                        | IntentStatus::RequiresCapture,
+                        SessionizerRefundStatus::NotRefunded,
+                        _,
+                    ) => sankey_response.smart_retried_success += i.count,
+                    (IntentStatus::Failed, _, 1) => sankey_response.normal_failure += i.count,
+                    (IntentStatus::Failed, _, _) => {
+                        sankey_response.smart_retried_failure += i.count
+                    }
+                    (IntentStatus::Cancelled, _, _) => sankey_response.cancelled += i.count,
+                    (IntentStatus::Processing, _, _) => sankey_response.pending += i.count,
+                    (IntentStatus::RequiresCustomerAction, _, _) => {
+                        sankey_response.customer_awaited += i.count
+                    }
+                    (IntentStatus::RequiresMerchantAction, _, _) => {
+                        sankey_response.merchant_awaited += i.count
+                    }
+                    (IntentStatus::RequiresPaymentMethod, _, _) => {
+                        sankey_response.pm_awaited += i.count
+                    }
+                    (IntentStatus::RequiresConfirmation, _, _) => {
+                        sankey_response.confirmation_awaited += i.count
+                    }
+                    i @ (_, _, _) => {
+                        router_env::logger::error!(status=?i, "Unknown status in sankey data");
+                    }
+                }
+            }
+            Ok(sankey_response)
+        }
+    }
+}
+
 #[instrument(skip_all)]
 pub async fn get_metrics(
     pool: &AnalyticsProvider,
     auth: &AuthInfo,
     req: GetPaymentIntentMetricRequest,
-) -> AnalyticsResult<MetricsResponse<MetricsBucketResponse>> {
+) -> AnalyticsResult<PaymentIntentsMetricsResponse<MetricsBucketResponse>> {
     let mut metrics_accumulator: HashMap<
         PaymentIntentMetricsBucketIdentifier,
         PaymentIntentMetricsAccumulator,
@@ -107,18 +183,34 @@ pub async fn get_metrics(
                     logger::debug!(bucket_id=?id, bucket_value=?value, "Bucket row for metric {metric}");
                     let metrics_builder = metrics_accumulator.entry(id).or_default();
                     match metric {
-                        PaymentIntentMetrics::SuccessfulSmartRetries => metrics_builder
-                            .successful_smart_retries
-                            .add_metrics_bucket(&value),
-                        PaymentIntentMetrics::TotalSmartRetries => metrics_builder
+                        PaymentIntentMetrics::SuccessfulSmartRetries
+                        | PaymentIntentMetrics::SessionizedSuccessfulSmartRetries => {
+                            metrics_builder
+                                .successful_smart_retries
+                                .add_metrics_bucket(&value)
+                        }
+                        PaymentIntentMetrics::TotalSmartRetries
+                        | PaymentIntentMetrics::SessionizedTotalSmartRetries => metrics_builder
                             .total_smart_retries
                             .add_metrics_bucket(&value),
-                        PaymentIntentMetrics::SmartRetriedAmount => metrics_builder
+                        PaymentIntentMetrics::SmartRetriedAmount
+                        | PaymentIntentMetrics::SessionizedSmartRetriedAmount => metrics_builder
                             .smart_retried_amount
                             .add_metrics_bucket(&value),
-                        PaymentIntentMetrics::PaymentIntentCount => metrics_builder
+                        PaymentIntentMetrics::PaymentIntentCount
+                        | PaymentIntentMetrics::SessionizedPaymentIntentCount => metrics_builder
                             .payment_intent_count
                             .add_metrics_bucket(&value),
+                        PaymentIntentMetrics::PaymentsSuccessRate
+                        | PaymentIntentMetrics::SessionizedPaymentsSuccessRate => metrics_builder
+                            .payments_success_rate
+                            .add_metrics_bucket(&value),
+                        PaymentIntentMetrics::SessionizedPaymentProcessedAmount => metrics_builder
+                            .payment_processed_amount
+                            .add_metrics_bucket(&value),
+                        PaymentIntentMetrics::SessionizedPaymentsDistribution => metrics_builder
+                            .payments_distribution
+                            .add_metrics_bucket(&value),
                     }
                 }
 
@@ -131,18 +223,80 @@ pub async fn get_metrics(
         }
     }
 
+    let mut success = 0;
+    let mut success_without_smart_retries = 0;
+    let mut total_smart_retried_amount = 0;
+    let mut total_smart_retried_amount_without_smart_retries = 0;
+    let mut total = 0;
+    let mut total_payment_processed_amount = 0;
+    let mut total_payment_processed_count = 0;
+    let mut total_payment_processed_amount_without_smart_retries = 0;
+    let mut total_payment_processed_count_without_smart_retries = 0;
     let query_data: Vec<MetricsBucketResponse> = metrics_accumulator
         .into_iter()
-        .map(|(id, val)| MetricsBucketResponse {
-            values: val.collect(),
-            dimensions: id,
+        .map(|(id, val)| {
+            let collected_values = val.collect();
+            if let Some(success_count) = collected_values.successful_payments {
+                success += success_count;
+            }
+            if let Some(success_count) = collected_values.successful_payments_without_smart_retries
+            {
+                success_without_smart_retries += success_count;
+            }
+            if let Some(total_count) = collected_values.total_payments {
+                total += total_count;
+            }
+            if let Some(retried_amount) = collected_values.smart_retried_amount {
+                total_smart_retried_amount += retried_amount;
+            }
+            if let Some(retried_amount) =
+                collected_values.smart_retried_amount_without_smart_retries
+            {
+                total_smart_retried_amount_without_smart_retries += retried_amount;
+            }
+            if let Some(amount) = collected_values.payment_processed_amount {
+                total_payment_processed_amount += amount;
+            }
+            if let Some(count) = collected_values.payment_processed_count {
+                total_payment_processed_count += count;
+            }
+            if let Some(amount) = collected_values.payment_processed_amount_without_smart_retries {
+                total_payment_processed_amount_without_smart_retries += amount;
+            }
+            if let Some(count) = collected_values.payment_processed_count_without_smart_retries {
+                total_payment_processed_count_without_smart_retries += count;
+            }
+            MetricsBucketResponse {
+                values: collected_values,
+                dimensions: id,
+            }
         })
         .collect();
-
-    Ok(MetricsResponse {
+    let total_success_rate = match (success, total) {
+        (s, t) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+        _ => None,
+    };
+    let total_success_rate_without_smart_retries = match (success_without_smart_retries, total) {
+        (s, t) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+        _ => None,
+    };
+    Ok(PaymentIntentsMetricsResponse {
         query_data,
-        meta_data: [AnalyticsMetadata {
-            current_time_range: req.time_range,
+        meta_data: [PaymentIntentsAnalyticsMetadata {
+            total_success_rate,
+            total_success_rate_without_smart_retries,
+            total_smart_retried_amount: Some(total_smart_retried_amount),
+            total_smart_retried_amount_without_smart_retries: Some(
+                total_smart_retried_amount_without_smart_retries,
+            ),
+            total_payment_processed_amount: Some(total_payment_processed_amount),
+            total_payment_processed_amount_without_smart_retries: Some(
+                total_payment_processed_amount_without_smart_retries,
+            ),
+            total_payment_processed_count: Some(total_payment_processed_count),
+            total_payment_processed_count_without_smart_retries: Some(
+                total_payment_processed_count_without_smart_retries,
+            ),
         }],
     })
 }
@@ -217,6 +371,15 @@ pub async fn get_filters(
             PaymentIntentDimensions::PaymentIntentStatus => fil.status.map(|i| i.as_ref().to_string()),
             PaymentIntentDimensions::Currency => fil.currency.map(|i| i.as_ref().to_string()),
             PaymentIntentDimensions::ProfileId => fil.profile_id,
+            PaymentIntentDimensions::Connector => fil.connector,
+            PaymentIntentDimensions::AuthType => fil.authentication_type.map(|i| i.as_ref().to_string()),
+            PaymentIntentDimensions::PaymentMethod => fil.payment_method,
+            PaymentIntentDimensions::PaymentMethodType => fil.payment_method_type,
+            PaymentIntentDimensions::CardNetwork => fil.card_network,
+            PaymentIntentDimensions::MerchantId => fil.merchant_id,
+            PaymentIntentDimensions::CardLast4 => fil.card_last_4,
+            PaymentIntentDimensions::CardIssuer => fil.card_issuer,
+            PaymentIntentDimensions::ErrorReason => fil.error_reason,
         })
         .collect::<Vec<String>>();
         res.query_data.push(PaymentIntentFilterValue {
diff --git a/crates/analytics/src/payment_intents/filters.rs b/crates/analytics/src/payment_intents/filters.rs
index e81b050214c..25d43e76f03 100644
--- a/crates/analytics/src/payment_intents/filters.rs
+++ b/crates/analytics/src/payment_intents/filters.rs
@@ -1,6 +1,6 @@
 use api_models::analytics::{payment_intents::PaymentIntentDimensions, Granularity, TimeRange};
 use common_utils::errors::ReportSwitchExt;
-use diesel_models::enums::{Currency, IntentStatus};
+use diesel_models::enums::{AuthenticationType, Currency, IntentStatus};
 use error_stack::ResultExt;
 use time::PrimitiveDateTime;
 
@@ -54,4 +54,13 @@ pub struct PaymentIntentFilterRow {
     pub status: Option<DBEnumWrapper<IntentStatus>>,
     pub currency: Option<DBEnumWrapper<Currency>>,
     pub profile_id: Option<String>,
+    pub connector: Option<String>,
+    pub authentication_type: Option<DBEnumWrapper<AuthenticationType>>,
+    pub payment_method: Option<String>,
+    pub payment_method_type: Option<String>,
+    pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
 }
diff --git a/crates/analytics/src/payment_intents/metrics.rs b/crates/analytics/src/payment_intents/metrics.rs
index f27a02368ae..8ee9d24b5a0 100644
--- a/crates/analytics/src/payment_intents/metrics.rs
+++ b/crates/analytics/src/payment_intents/metrics.rs
@@ -17,11 +17,14 @@ use crate::{
 };
 
 mod payment_intent_count;
+mod payments_success_rate;
+mod sessionized_metrics;
 mod smart_retried_amount;
 mod successful_smart_retries;
 mod total_smart_retries;
 
 use payment_intent_count::PaymentIntentCount;
+use payments_success_rate::PaymentsSuccessRate;
 use smart_retried_amount::SmartRetriedAmount;
 use successful_smart_retries::SuccessfulSmartRetries;
 use total_smart_retries::TotalSmartRetries;
@@ -31,6 +34,16 @@ pub struct PaymentIntentMetricRow {
     pub status: Option<DBEnumWrapper<storage_enums::IntentStatus>>,
     pub currency: Option<DBEnumWrapper<storage_enums::Currency>>,
     pub profile_id: Option<String>,
+    pub connector: Option<String>,
+    pub authentication_type: Option<DBEnumWrapper<storage_enums::AuthenticationType>>,
+    pub payment_method: Option<String>,
+    pub payment_method_type: Option<String>,
+    pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
+    pub first_attempt: Option<i64>,
     pub total: Option<bigdecimal::BigDecimal>,
     pub count: Option<i64>,
     #[serde(with = "common_utils::custom_serde::iso8601::option")]
@@ -98,6 +111,46 @@ where
                     .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
                     .await
             }
+            Self::PaymentsSuccessRate => {
+                PaymentsSuccessRate
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedSuccessfulSmartRetries => {
+                sessionized_metrics::SuccessfulSmartRetries
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedTotalSmartRetries => {
+                sessionized_metrics::TotalSmartRetries
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedSmartRetriedAmount => {
+                sessionized_metrics::SmartRetriedAmount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedPaymentIntentCount => {
+                sessionized_metrics::PaymentIntentCount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedPaymentsSuccessRate => {
+                sessionized_metrics::PaymentsSuccessRate
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedPaymentProcessedAmount => {
+                sessionized_metrics::PaymentProcessedAmount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedPaymentsDistribution => {
+                sessionized_metrics::PaymentsDistribution
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
         }
     }
 }
diff --git a/crates/analytics/src/payment_intents/metrics/payment_intent_count.rs b/crates/analytics/src/payment_intents/metrics/payment_intent_count.rs
index 4632cbe9f37..b301a9b9b23 100644
--- a/crates/analytics/src/payment_intents/metrics/payment_intent_count.rs
+++ b/crates/analytics/src/payment_intents/metrics/payment_intent_count.rs
@@ -101,6 +101,15 @@ where
                         i.status.as_ref().map(|i| i.0),
                         i.currency.as_ref().map(|i| i.0),
                         i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payment_intents/metrics/payments_success_rate.rs b/crates/analytics/src/payment_intents/metrics/payments_success_rate.rs
new file mode 100644
index 00000000000..07b1bfcf69f
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/payments_success_rate.rs
@@ -0,0 +1,146 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(super) struct PaymentsSuccessRate;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for PaymentsSuccessRate
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntent);
+
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentIntentDimensions::PaymentIntentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column("(attempt_count == 1) as first_attempt".to_string())
+            .switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        None,
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/mod.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/mod.rs
new file mode 100644
index 00000000000..fcd9a2b8adf
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/mod.rs
@@ -0,0 +1,17 @@
+mod payment_intent_count;
+mod payment_processed_amount;
+mod payments_distribution;
+mod payments_success_rate;
+mod smart_retried_amount;
+mod successful_smart_retries;
+mod total_smart_retries;
+
+pub(super) use payment_intent_count::PaymentIntentCount;
+pub(super) use payment_processed_amount::PaymentProcessedAmount;
+pub(super) use payments_distribution::PaymentsDistribution;
+pub(super) use payments_success_rate::PaymentsSuccessRate;
+pub(super) use smart_retried_amount::SmartRetriedAmount;
+pub(super) use successful_smart_retries::SuccessfulSmartRetries;
+pub(super) use total_smart_retries::TotalSmartRetries;
+
+pub use super::{PaymentIntentMetric, PaymentIntentMetricAnalytics, PaymentIntentMetricRow};
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payment_intent_count.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payment_intent_count.rs
new file mode 100644
index 00000000000..7475a75bb53
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payment_intent_count.rs
@@ -0,0 +1,133 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentIntentCount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for PaymentIntentCount
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        i.status.as_ref().map(|i| i.0),
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payment_processed_amount.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payment_processed_amount.rs
new file mode 100644
index 00000000000..506965375f5
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payment_processed_amount.rs
@@ -0,0 +1,160 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentProcessedAmount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for PaymentProcessedAmount
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentIntentDimensions::PaymentIntentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column("attempt_count == 1 as first_attempt")
+            .switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Sum {
+                field: "amount",
+                alias: Some("total"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        query_builder
+            .add_group_by_clause("attempt_count")
+            .attach_printable("Error grouping by attempt_count")
+            .switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .add_filter_clause(
+                PaymentIntentDimensions::PaymentIntentStatus,
+                storage_enums::IntentStatus::Succeeded,
+            )
+            .switch()?;
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        None,
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payments_distribution.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payments_distribution.rs
new file mode 100644
index 00000000000..0b55c101a7c
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payments_distribution.rs
@@ -0,0 +1,145 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentsDistribution;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for PaymentsDistribution
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentIntentDimensions::PaymentIntentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column("attempt_count == 1 as first_attempt")
+            .switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+        query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        None,
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payments_success_rate.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payments_success_rate.rs
new file mode 100644
index 00000000000..8c340d0b2d6
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/payments_success_rate.rs
@@ -0,0 +1,146 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentsSuccessRate;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for PaymentsSuccessRate
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentIntentDimensions::PaymentIntentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column("(attempt_count == 1) as first_attempt".to_string())
+            .switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        None,
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/smart_retried_amount.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/smart_retried_amount.rs
new file mode 100644
index 00000000000..8105a4c82a4
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/smart_retried_amount.rs
@@ -0,0 +1,154 @@
+use std::collections::HashSet;
+
+use api_models::{
+    analytics::{
+        payment_intents::{
+            PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+        },
+        Granularity, TimeRange,
+    },
+    enums::IntentStatus,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct SmartRetriedAmount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for SmartRetriedAmount
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+        query_builder
+            .add_select_column(Aggregate::Sum {
+                field: "amount",
+                alias: Some("total"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_select_column("attempt_count == 1 as first_attempt")
+            .switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        query_builder
+            .add_custom_filter_clause("attempt_count", "1", FilterTypes::Gt)
+            .switch()?;
+        query_builder
+            .add_custom_filter_clause("status", IntentStatus::Succeeded, FilterTypes::Equal)
+            .switch()?;
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        i.status.as_ref().map(|i| i.0),
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/successful_smart_retries.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/successful_smart_retries.rs
new file mode 100644
index 00000000000..0b28cb5366d
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/successful_smart_retries.rs
@@ -0,0 +1,143 @@
+use std::collections::HashSet;
+
+use api_models::{
+    analytics::{
+        payment_intents::{
+            PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+        },
+        Granularity, TimeRange,
+    },
+    enums::IntentStatus,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct SuccessfulSmartRetries;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for SuccessfulSmartRetries
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        query_builder
+            .add_custom_filter_clause("attempt_count", "1", FilterTypes::Gt)
+            .switch()?;
+        query_builder
+            .add_custom_filter_clause("status", IntentStatus::Succeeded, FilterTypes::Equal)
+            .switch()?;
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        i.status.as_ref().map(|i| i.0),
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/sessionized_metrics/total_smart_retries.rs b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/total_smart_retries.rs
new file mode 100644
index 00000000000..20ef8be6277
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/sessionized_metrics/total_smart_retries.rs
@@ -0,0 +1,138 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentIntentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct TotalSmartRetries;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentIntentMetric<T> for TotalSmartRetries
+where
+    T: AnalyticsDataSource + super::PaymentIntentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        query_builder
+            .add_custom_filter_clause("attempt_count", "1", FilterTypes::Gt)
+            .switch()?;
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentIntentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentIntentMetricsBucketIdentifier::new(
+                        i.status.as_ref().map(|i| i.0),
+                        i.currency.as_ref().map(|i| i.0),
+                        i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/smart_retried_amount.rs b/crates/analytics/src/payment_intents/metrics/smart_retried_amount.rs
index e1df9fe50d3..8468911f7bb 100644
--- a/crates/analytics/src/payment_intents/metrics/smart_retried_amount.rs
+++ b/crates/analytics/src/payment_intents/metrics/smart_retried_amount.rs
@@ -58,6 +58,11 @@ where
                 alias: Some("total"),
             })
             .switch()?;
+
+        query_builder
+            .add_select_column("attempt_count == 1 as first_attempt")
+            .switch()?;
+
         query_builder
             .add_select_column(Aggregate::Min {
                 field: "created_at",
@@ -93,6 +98,11 @@ where
                 .switch()?;
         }
 
+        query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+
         if let Some(granularity) = granularity.as_ref() {
             granularity
                 .set_group_by_clause(&mut query_builder)
@@ -112,6 +122,15 @@ where
                         i.status.as_ref().map(|i| i.0),
                         i.currency.as_ref().map(|i| i.0),
                         i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payment_intents/metrics/successful_smart_retries.rs b/crates/analytics/src/payment_intents/metrics/successful_smart_retries.rs
index 4fe5f3a26f5..a19bdec518c 100644
--- a/crates/analytics/src/payment_intents/metrics/successful_smart_retries.rs
+++ b/crates/analytics/src/payment_intents/metrics/successful_smart_retries.rs
@@ -111,6 +111,15 @@ where
                         i.status.as_ref().map(|i| i.0),
                         i.currency.as_ref().map(|i| i.0),
                         i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payment_intents/metrics/total_smart_retries.rs b/crates/analytics/src/payment_intents/metrics/total_smart_retries.rs
index e98efa9f6ab..f5539abd9f5 100644
--- a/crates/analytics/src/payment_intents/metrics/total_smart_retries.rs
+++ b/crates/analytics/src/payment_intents/metrics/total_smart_retries.rs
@@ -106,6 +106,15 @@ where
                         i.status.as_ref().map(|i| i.0),
                         i.currency.as_ref().map(|i| i.0),
                         i.profile_id.clone(),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payment_intents/sankey.rs b/crates/analytics/src/payment_intents/sankey.rs
new file mode 100644
index 00000000000..53fd03562f1
--- /dev/null
+++ b/crates/analytics/src/payment_intents/sankey.rs
@@ -0,0 +1,127 @@
+use common_enums::enums;
+use common_utils::{
+    errors::ParsingError,
+    types::{authentication::AuthInfo, TimeRange},
+};
+use error_stack::ResultExt;
+use router_env::logger;
+use time::PrimitiveDateTime;
+
+use crate::{
+    clickhouse::ClickhouseClient,
+    query::{Aggregate, QueryBuilder, QueryFilter},
+    types::{AnalyticsCollection, DBEnumWrapper, MetricsError, MetricsResult},
+};
+
+#[derive(Debug, PartialEq, Eq, serde::Deserialize, Hash)]
+pub struct PaymentIntentMetricRow {
+    pub profile_id: Option<String>,
+    pub connector: Option<String>,
+    pub authentication_type: Option<DBEnumWrapper<enums::AuthenticationType>>,
+    pub payment_method: Option<String>,
+    pub payment_method_type: Option<String>,
+    pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
+    pub first_attempt: Option<i64>,
+    pub total: Option<bigdecimal::BigDecimal>,
+    pub count: Option<i64>,
+    #[serde(with = "common_utils::custom_serde::iso8601::option")]
+    pub start_bucket: Option<PrimitiveDateTime>,
+    #[serde(with = "common_utils::custom_serde::iso8601::option")]
+    pub end_bucket: Option<PrimitiveDateTime>,
+}
+
+#[derive(
+    Debug, Default, serde::Deserialize, strum::AsRefStr, strum::EnumString, strum::Display,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum SessionizerRefundStatus {
+    FullRefunded,
+    #[default]
+    NotRefunded,
+    PartialRefunded,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct SankeyRow {
+    pub status: DBEnumWrapper<enums::IntentStatus>,
+    #[serde(default)]
+    pub refunds_status: Option<DBEnumWrapper<SessionizerRefundStatus>>,
+    pub attempt_count: i64,
+    pub count: i64,
+}
+
+impl TryInto<SankeyRow> for serde_json::Value {
+    type Error = error_stack::Report<ParsingError>;
+
+    fn try_into(self) -> Result<SankeyRow, Self::Error> {
+        logger::debug!("Parsing SankeyRow from {:?}", self);
+        serde_json::from_value(self).change_context(ParsingError::StructParseFailure(
+            "Failed to parse Sankey in clickhouse results",
+        ))
+    }
+}
+
+pub async fn get_sankey_data(
+    clickhouse_client: &ClickhouseClient,
+    auth: &AuthInfo,
+    time_range: &TimeRange,
+) -> MetricsResult<Vec<SankeyRow>> {
+    let mut query_builder =
+        QueryBuilder::<ClickhouseClient>::new(AnalyticsCollection::PaymentIntentSessionized);
+    query_builder
+        .add_select_column(Aggregate::<String>::Count {
+            field: None,
+            alias: Some("count"),
+        })
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    query_builder
+        .add_select_column("status")
+        .attach_printable("Error adding select clause")
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    query_builder
+        .add_select_column("refunds_status")
+        .attach_printable("Error adding select clause")
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    query_builder
+        .add_select_column("attempt_count")
+        .attach_printable("Error adding select clause")
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    auth.set_filter_clause(&mut query_builder)
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    time_range
+        .set_filter_clause(&mut query_builder)
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    query_builder
+        .add_group_by_clause("status")
+        .attach_printable("Error adding group by clause")
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    query_builder
+        .add_group_by_clause("refunds_status")
+        .attach_printable("Error adding group by clause")
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    query_builder
+        .add_group_by_clause("attempt_count")
+        .attach_printable("Error adding group by clause")
+        .change_context(MetricsError::QueryBuildingError)?;
+
+    query_builder
+        .execute_query::<SankeyRow, _>(clickhouse_client)
+        .await
+        .change_context(MetricsError::QueryBuildingError)?
+        .change_context(MetricsError::QueryExecutionFailure)?
+        .into_iter()
+        .map(Ok)
+        .collect()
+}
diff --git a/crates/analytics/src/payment_intents/types.rs b/crates/analytics/src/payment_intents/types.rs
index 03f2a196c20..1a9a2f2ed65 100644
--- a/crates/analytics/src/payment_intents/types.rs
+++ b/crates/analytics/src/payment_intents/types.rs
@@ -30,6 +30,63 @@ where
                 .add_filter_in_range_clause(PaymentIntentDimensions::ProfileId, &self.profile_id)
                 .attach_printable("Error adding profile id filter")?;
         }
+        if !self.connector.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentIntentDimensions::Connector, &self.connector)
+                .attach_printable("Error adding connector filter")?;
+        }
+        if !self.auth_type.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentIntentDimensions::AuthType, &self.auth_type)
+                .attach_printable("Error adding auth type filter")?;
+        }
+        if !self.payment_method.is_empty() {
+            builder
+                .add_filter_in_range_clause(
+                    PaymentIntentDimensions::PaymentMethod,
+                    &self.payment_method,
+                )
+                .attach_printable("Error adding payment method filter")?;
+        }
+        if !self.payment_method_type.is_empty() {
+            builder
+                .add_filter_in_range_clause(
+                    PaymentIntentDimensions::PaymentMethodType,
+                    &self.payment_method_type,
+                )
+                .attach_printable("Error adding payment method type filter")?;
+        }
+        if !self.card_network.is_empty() {
+            builder
+                .add_filter_in_range_clause(
+                    PaymentIntentDimensions::CardNetwork,
+                    &self.card_network,
+                )
+                .attach_printable("Error adding card network filter")?;
+        }
+        if !self.merchant_id.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentIntentDimensions::MerchantId, &self.merchant_id)
+                .attach_printable("Error adding merchant id filter")?;
+        }
+        if !self.card_last_4.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentIntentDimensions::CardLast4, &self.card_last_4)
+                .attach_printable("Error adding card last 4 filter")?;
+        }
+        if !self.card_issuer.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentIntentDimensions::CardIssuer, &self.card_issuer)
+                .attach_printable("Error adding card issuer filter")?;
+        }
+        if !self.error_reason.is_empty() {
+            builder
+                .add_filter_in_range_clause(
+                    PaymentIntentDimensions::ErrorReason,
+                    &self.error_reason,
+                )
+                .attach_printable("Error adding error reason filter")?;
+        }
         Ok(())
     }
 }
diff --git a/crates/analytics/src/payments/accumulator.rs b/crates/analytics/src/payments/accumulator.rs
index efc8aaf6983..4388b2071fe 100644
--- a/crates/analytics/src/payments/accumulator.rs
+++ b/crates/analytics/src/payments/accumulator.rs
@@ -10,12 +10,14 @@ pub struct PaymentMetricsAccumulator {
     pub payment_success_rate: SuccessRateAccumulator,
     pub payment_count: CountAccumulator,
     pub payment_success: CountAccumulator,
-    pub processed_amount: SumAccumulator,
+    pub processed_amount: ProcessedAmountAccumulator,
     pub avg_ticket_size: AverageAccumulator,
     pub payment_error_message: ErrorDistributionAccumulator,
     pub retries_count: CountAccumulator,
-    pub retries_amount_processed: SumAccumulator,
+    pub retries_amount_processed: RetriesAmountAccumulator,
     pub connector_success_rate: SuccessRateAccumulator,
+    pub payments_distribution: PaymentsDistributionAccumulator,
+    pub failure_reasons_distribution: FailureReasonsDistributionAccumulator,
 }
 
 #[derive(Debug, Default)]
@@ -30,6 +32,12 @@ pub struct ErrorDistributionAccumulator {
     pub error_vec: Vec<ErrorDistributionRow>,
 }
 
+#[derive(Debug, Default)]
+pub struct FailureReasonsDistributionAccumulator {
+    pub count: u64,
+    pub count_without_retries: u64,
+}
+
 #[derive(Debug, Default)]
 pub struct SuccessRateAccumulator {
     pub success: i64,
@@ -43,9 +51,11 @@ pub struct CountAccumulator {
 }
 
 #[derive(Debug, Default)]
-#[repr(transparent)]
-pub struct SumAccumulator {
-    pub total: Option<i64>,
+pub struct ProcessedAmountAccumulator {
+    pub count_with_retries: Option<i64>,
+    pub total_with_retries: Option<i64>,
+    pub count_without_retries: Option<i64>,
+    pub total_without_retries: Option<i64>,
 }
 
 #[derive(Debug, Default)]
@@ -54,6 +64,22 @@ pub struct AverageAccumulator {
     pub count: u32,
 }
 
+#[derive(Debug, Default)]
+#[repr(transparent)]
+pub struct RetriesAmountAccumulator {
+    pub total: Option<i64>,
+}
+
+#[derive(Debug, Default)]
+pub struct PaymentsDistributionAccumulator {
+    pub success: u32,
+    pub failed: u32,
+    pub total: u32,
+    pub success_without_retries: u32,
+    pub failed_without_retries: u32,
+    pub total_without_retries: u32,
+}
+
 pub trait PaymentMetricAccumulator {
     type MetricOutput;
 
@@ -107,6 +133,29 @@ impl PaymentDistributionAccumulator for ErrorDistributionAccumulator {
     }
 }
 
+impl PaymentMetricAccumulator for FailureReasonsDistributionAccumulator {
+    type MetricOutput = (Option<u64>, Option<u64>);
+
+    fn add_metrics_bucket(&mut self, metrics: &PaymentMetricRow) {
+        if let Some(count) = metrics.count {
+            if let Ok(count_u64) = u64::try_from(count) {
+                self.count += count_u64;
+            }
+        }
+        if metrics.first_attempt.unwrap_or(false) {
+            if let Some(count) = metrics.count {
+                if let Ok(count_u64) = u64::try_from(count) {
+                    self.count_without_retries += count_u64;
+                }
+            }
+        }
+    }
+
+    fn collect(self) -> Self::MetricOutput {
+        (Some(self.count), Some(self.count_without_retries))
+    }
+}
+
 impl PaymentMetricAccumulator for SuccessRateAccumulator {
     type MetricOutput = Option<f64>;
 
@@ -131,6 +180,81 @@ impl PaymentMetricAccumulator for SuccessRateAccumulator {
     }
 }
 
+impl PaymentMetricAccumulator for PaymentsDistributionAccumulator {
+    type MetricOutput = (Option<f64>, Option<f64>, Option<f64>, Option<f64>);
+
+    fn add_metrics_bucket(&mut self, metrics: &PaymentMetricRow) {
+        if let Some(ref status) = metrics.status {
+            if status.as_ref() == &storage_enums::AttemptStatus::Charged {
+                if let Some(success) = metrics
+                    .count
+                    .and_then(|success| u32::try_from(success).ok())
+                {
+                    self.success += success;
+                    if metrics.first_attempt.unwrap_or(false) {
+                        self.success_without_retries += success;
+                    }
+                }
+            }
+            if status.as_ref() == &storage_enums::AttemptStatus::Failure {
+                if let Some(failed) = metrics.count.and_then(|failed| u32::try_from(failed).ok()) {
+                    self.failed += failed;
+                    if metrics.first_attempt.unwrap_or(false) {
+                        self.failed_without_retries += failed;
+                    }
+                }
+            }
+            if let Some(total) = metrics.count.and_then(|total| u32::try_from(total).ok()) {
+                self.total += total;
+                if metrics.first_attempt.unwrap_or(false) {
+                    self.total_without_retries += total;
+                }
+            }
+        }
+    }
+
+    fn collect(self) -> Self::MetricOutput {
+        if self.total == 0 {
+            (None, None, None, None)
+        } else {
+            let success = Some(self.success);
+            let success_without_retries = Some(self.success_without_retries);
+            let failed = Some(self.failed);
+            let failed_without_retries = Some(self.failed_without_retries);
+            let total = Some(self.total);
+            let total_without_retries = Some(self.total_without_retries);
+
+            let success_rate = match (success, total) {
+                (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                _ => None,
+            };
+
+            let success_rate_without_retries =
+                match (success_without_retries, total_without_retries) {
+                    (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                    _ => None,
+                };
+
+            let failed_rate = match (failed, total) {
+                (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                _ => None,
+            };
+
+            let failed_rate_without_retries = match (failed_without_retries, total_without_retries)
+            {
+                (Some(s), Some(t)) if t > 0 => Some(f64::from(s) * 100.0 / f64::from(t)),
+                _ => None,
+            };
+            (
+                success_rate,
+                success_rate_without_retries,
+                failed_rate,
+                failed_rate_without_retries,
+            )
+        }
+    }
+}
+
 impl PaymentMetricAccumulator for CountAccumulator {
     type MetricOutput = Option<u64>;
     #[inline]
@@ -147,9 +271,63 @@ impl PaymentMetricAccumulator for CountAccumulator {
     }
 }
 
-impl PaymentMetricAccumulator for SumAccumulator {
-    type MetricOutput = Option<u64>;
+impl PaymentMetricAccumulator for ProcessedAmountAccumulator {
+    type MetricOutput = (Option<u64>, Option<u64>, Option<u64>, Option<u64>);
     #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &PaymentMetricRow) {
+        self.total_with_retries = match (
+            self.total_with_retries,
+            metrics.total.as_ref().and_then(ToPrimitive::to_i64),
+        ) {
+            (None, None) => None,
+            (None, i @ Some(_)) | (i @ Some(_), None) => i,
+            (Some(a), Some(b)) => Some(a + b),
+        };
+
+        self.count_with_retries = match (self.count_with_retries, metrics.count) {
+            (None, None) => None,
+            (None, i @ Some(_)) | (i @ Some(_), None) => i,
+            (Some(a), Some(b)) => Some(a + b),
+        };
+
+        if metrics.first_attempt.unwrap_or(false) {
+            self.total_without_retries = match (
+                self.total_without_retries,
+                metrics.total.as_ref().and_then(ToPrimitive::to_i64),
+            ) {
+                (None, None) => None,
+                (None, i @ Some(_)) | (i @ Some(_), None) => i,
+                (Some(a), Some(b)) => Some(a + b),
+            };
+
+            self.count_without_retries = match (self.count_without_retries, metrics.count) {
+                (None, None) => None,
+                (None, i @ Some(_)) | (i @ Some(_), None) => i,
+                (Some(a), Some(b)) => Some(a + b),
+            };
+        }
+    }
+    #[inline]
+    fn collect(self) -> Self::MetricOutput {
+        let total_with_retries = u64::try_from(self.total_with_retries.unwrap_or(0)).ok();
+        let count_with_retries = self.count_with_retries.and_then(|i| u64::try_from(i).ok());
+
+        let total_without_retries = u64::try_from(self.total_without_retries.unwrap_or(0)).ok();
+        let count_without_retries = self
+            .count_without_retries
+            .and_then(|i| u64::try_from(i).ok());
+
+        (
+            total_with_retries,
+            count_with_retries,
+            total_without_retries,
+            count_without_retries,
+        )
+    }
+}
+
+impl PaymentMetricAccumulator for RetriesAmountAccumulator {
+    type MetricOutput = Option<u64>;
     fn add_metrics_bucket(&mut self, metrics: &PaymentMetricRow) {
         self.total = match (
             self.total,
@@ -158,7 +336,7 @@ impl PaymentMetricAccumulator for SumAccumulator {
             (None, None) => None,
             (None, i @ Some(_)) | (i @ Some(_), None) => i,
             (Some(a), Some(b)) => Some(a + b),
-        }
+        };
     }
     #[inline]
     fn collect(self) -> Self::MetricOutput {
@@ -195,16 +373,39 @@ impl PaymentMetricAccumulator for AverageAccumulator {
 
 impl PaymentMetricsAccumulator {
     pub fn collect(self) -> PaymentMetricsBucketValue {
+        let (
+            payment_processed_amount,
+            payment_processed_count,
+            payment_processed_amount_without_smart_retries,
+            payment_processed_count_without_smart_retries,
+        ) = self.processed_amount.collect();
+        let (
+            payments_success_rate_distribution,
+            payments_success_rate_distribution_without_smart_retries,
+            payments_failure_rate_distribution,
+            payments_failure_rate_distribution_without_smart_retries,
+        ) = self.payments_distribution.collect();
+        let (failure_reason_count, failure_reason_count_without_smart_retries) =
+            self.failure_reasons_distribution.collect();
         PaymentMetricsBucketValue {
             payment_success_rate: self.payment_success_rate.collect(),
             payment_count: self.payment_count.collect(),
             payment_success_count: self.payment_success.collect(),
-            payment_processed_amount: self.processed_amount.collect(),
+            payment_processed_amount,
+            payment_processed_count,
+            payment_processed_amount_without_smart_retries,
+            payment_processed_count_without_smart_retries,
             avg_ticket_size: self.avg_ticket_size.collect(),
             payment_error_message: self.payment_error_message.collect(),
             retries_count: self.retries_count.collect(),
             retries_amount_processed: self.retries_amount_processed.collect(),
             connector_success_rate: self.connector_success_rate.collect(),
+            payments_success_rate_distribution,
+            payments_success_rate_distribution_without_smart_retries,
+            payments_failure_rate_distribution,
+            payments_failure_rate_distribution_without_smart_retries,
+            failure_reason_count,
+            failure_reason_count_without_smart_retries,
         }
     }
 }
diff --git a/crates/analytics/src/payments/core.rs b/crates/analytics/src/payments/core.rs
index f41e1719450..59ae549b283 100644
--- a/crates/analytics/src/payments/core.rs
+++ b/crates/analytics/src/payments/core.rs
@@ -6,8 +6,8 @@ use api_models::analytics::{
         MetricsBucketResponse, PaymentDimensions, PaymentDistributions, PaymentMetrics,
         PaymentMetricsBucketIdentifier,
     },
-    AnalyticsMetadata, FilterValue, GetPaymentFiltersRequest, GetPaymentMetricRequest,
-    MetricsResponse, PaymentFiltersResponse,
+    FilterValue, GetPaymentFiltersRequest, GetPaymentMetricRequest, PaymentFiltersResponse,
+    PaymentsAnalyticsMetadata, PaymentsMetricsResponse,
 };
 use common_utils::errors::CustomResult;
 use error_stack::ResultExt;
@@ -48,7 +48,7 @@ pub async fn get_metrics(
     pool: &AnalyticsProvider,
     auth: &AuthInfo,
     req: GetPaymentMetricRequest,
-) -> AnalyticsResult<MetricsResponse<MetricsBucketResponse>> {
+) -> AnalyticsResult<PaymentsMetricsResponse<MetricsBucketResponse>> {
     let mut metrics_accumulator: HashMap<
         PaymentMetricsBucketIdentifier,
         PaymentMetricsAccumulator,
@@ -137,32 +137,47 @@ pub async fn get_metrics(
                     logger::debug!(bucket_id=?id, bucket_value=?value, "Bucket row for metric {metric}");
                     let metrics_builder = metrics_accumulator.entry(id).or_default();
                     match metric {
-                        PaymentMetrics::PaymentSuccessRate => metrics_builder
+                        PaymentMetrics::PaymentSuccessRate
+                        | PaymentMetrics::SessionizedPaymentSuccessRate => metrics_builder
                             .payment_success_rate
                             .add_metrics_bucket(&value),
-                        PaymentMetrics::PaymentCount => {
+                        PaymentMetrics::PaymentCount | PaymentMetrics::SessionizedPaymentCount => {
                             metrics_builder.payment_count.add_metrics_bucket(&value)
                         }
-                        PaymentMetrics::PaymentSuccessCount => {
+                        PaymentMetrics::PaymentSuccessCount
+                        | PaymentMetrics::SessionizedPaymentSuccessCount => {
                             metrics_builder.payment_success.add_metrics_bucket(&value)
                         }
-                        PaymentMetrics::PaymentProcessedAmount => {
+                        PaymentMetrics::PaymentProcessedAmount
+                        | PaymentMetrics::SessionizedPaymentProcessedAmount => {
                             metrics_builder.processed_amount.add_metrics_bucket(&value)
                         }
-                        PaymentMetrics::AvgTicketSize => {
+                        PaymentMetrics::AvgTicketSize
+                        | PaymentMetrics::SessionizedAvgTicketSize => {
                             metrics_builder.avg_ticket_size.add_metrics_bucket(&value)
                         }
-                        PaymentMetrics::RetriesCount => {
+                        PaymentMetrics::RetriesCount | PaymentMetrics::SessionizedRetriesCount => {
                             metrics_builder.retries_count.add_metrics_bucket(&value);
                             metrics_builder
                                 .retries_amount_processed
                                 .add_metrics_bucket(&value)
                         }
-                        PaymentMetrics::ConnectorSuccessRate => {
+                        PaymentMetrics::ConnectorSuccessRate
+                        | PaymentMetrics::SessionizedConnectorSuccessRate => {
                             metrics_builder
                                 .connector_success_rate
                                 .add_metrics_bucket(&value);
                         }
+                        PaymentMetrics::PaymentsDistribution => {
+                            metrics_builder
+                                .payments_distribution
+                                .add_metrics_bucket(&value);
+                        }
+                        PaymentMetrics::FailureReasons => {
+                            metrics_builder
+                                .failure_reasons_distribution
+                                .add_metrics_bucket(&value);
+                        }
                     }
                 }
 
@@ -203,19 +218,56 @@ pub async fn get_metrics(
             }
         }
     }
-
+    let mut total_payment_processed_amount = 0;
+    let mut total_payment_processed_count = 0;
+    let mut total_payment_processed_amount_without_smart_retries = 0;
+    let mut total_payment_processed_count_without_smart_retries = 0;
+    let mut total_failure_reasons_count = 0;
+    let mut total_failure_reasons_count_without_smart_retries = 0;
     let query_data: Vec<MetricsBucketResponse> = metrics_accumulator
         .into_iter()
-        .map(|(id, val)| MetricsBucketResponse {
-            values: val.collect(),
-            dimensions: id,
+        .map(|(id, val)| {
+            let collected_values = val.collect();
+            if let Some(amount) = collected_values.payment_processed_amount {
+                total_payment_processed_amount += amount;
+            }
+            if let Some(count) = collected_values.payment_processed_count {
+                total_payment_processed_count += count;
+            }
+            if let Some(amount) = collected_values.payment_processed_amount_without_smart_retries {
+                total_payment_processed_amount_without_smart_retries += amount;
+            }
+            if let Some(count) = collected_values.payment_processed_count_without_smart_retries {
+                total_payment_processed_count_without_smart_retries += count;
+            }
+            if let Some(count) = collected_values.failure_reason_count {
+                total_failure_reasons_count += count;
+            }
+            if let Some(count) = collected_values.failure_reason_count_without_smart_retries {
+                total_failure_reasons_count_without_smart_retries += count;
+            }
+            MetricsBucketResponse {
+                values: collected_values,
+                dimensions: id,
+            }
         })
         .collect();
 
-    Ok(MetricsResponse {
+    Ok(PaymentsMetricsResponse {
         query_data,
-        meta_data: [AnalyticsMetadata {
-            current_time_range: req.time_range,
+        meta_data: [PaymentsAnalyticsMetadata {
+            total_payment_processed_amount: Some(total_payment_processed_amount),
+            total_payment_processed_amount_without_smart_retries: Some(
+                total_payment_processed_amount_without_smart_retries,
+            ),
+            total_payment_processed_count: Some(total_payment_processed_count),
+            total_payment_processed_count_without_smart_retries: Some(
+                total_payment_processed_count_without_smart_retries,
+            ),
+            total_failure_reasons_count: Some(total_failure_reasons_count),
+            total_failure_reasons_count_without_smart_retries: Some(
+                total_failure_reasons_count_without_smart_retries,
+            ),
         }],
     })
 }
@@ -297,6 +349,10 @@ pub async fn get_filters(
             PaymentDimensions::ClientVersion => fil.client_version,
             PaymentDimensions::ProfileId => fil.profile_id,
             PaymentDimensions::CardNetwork => fil.card_network,
+            PaymentDimensions::MerchantId => fil.merchant_id,
+            PaymentDimensions::CardLast4 => fil.card_last_4,
+            PaymentDimensions::CardIssuer => fil.card_issuer,
+            PaymentDimensions::ErrorReason => fil.error_reason,
         })
         .collect::<Vec<String>>();
         res.query_data.push(FilterValue {
diff --git a/crates/analytics/src/payments/distribution.rs b/crates/analytics/src/payments/distribution.rs
index 3bd8e5f7815..213b6244574 100644
--- a/crates/analytics/src/payments/distribution.rs
+++ b/crates/analytics/src/payments/distribution.rs
@@ -28,6 +28,12 @@ pub struct PaymentDistributionRow {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub profile_id: Option<String>,
+    pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
+    pub first_attempt: Option<bool>,
     pub total: Option<bigdecimal::BigDecimal>,
     pub count: Option<i64>,
     pub error_message: Option<String>,
diff --git a/crates/analytics/src/payments/distribution/payment_error_message.rs b/crates/analytics/src/payments/distribution/payment_error_message.rs
index 087ec5a640e..241754ee041 100644
--- a/crates/analytics/src/payments/distribution/payment_error_message.rs
+++ b/crates/analytics/src/payments/distribution/payment_error_message.rs
@@ -155,6 +155,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/filters.rs b/crates/analytics/src/payments/filters.rs
index 094fd574169..51805acaae2 100644
--- a/crates/analytics/src/payments/filters.rs
+++ b/crates/analytics/src/payments/filters.rs
@@ -60,4 +60,8 @@ pub struct PaymentFilterRow {
     pub client_version: Option<String>,
     pub profile_id: Option<String>,
     pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
 }
diff --git a/crates/analytics/src/payments/metrics.rs b/crates/analytics/src/payments/metrics.rs
index 731f03a247a..23b133ad035 100644
--- a/crates/analytics/src/payments/metrics.rs
+++ b/crates/analytics/src/payments/metrics.rs
@@ -19,6 +19,7 @@ mod payment_count;
 mod payment_processed_amount;
 mod payment_success_count;
 mod retries_count;
+mod sessionized_metrics;
 mod success_rate;
 
 use avg_ticket_size::AvgTicketSize;
@@ -41,6 +42,12 @@ pub struct PaymentMetricRow {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub profile_id: Option<String>,
+    pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
+    pub first_attempt: Option<bool>,
     pub total: Option<bigdecimal::BigDecimal>,
     pub count: Option<i64>,
     #[serde(with = "common_utils::custom_serde::iso8601::option")]
@@ -122,6 +129,51 @@ where
                     .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
                     .await
             }
+            Self::SessionizedPaymentSuccessRate => {
+                sessionized_metrics::PaymentSuccessRate
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedPaymentCount => {
+                sessionized_metrics::PaymentCount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedPaymentSuccessCount => {
+                sessionized_metrics::PaymentSuccessCount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedPaymentProcessedAmount => {
+                sessionized_metrics::PaymentProcessedAmount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedAvgTicketSize => {
+                sessionized_metrics::AvgTicketSize
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedRetriesCount => {
+                sessionized_metrics::RetriesCount
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedConnectorSuccessRate => {
+                sessionized_metrics::ConnectorSuccessRate
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::PaymentsDistribution => {
+                sessionized_metrics::PaymentsDistribution
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::FailureReasons => {
+                sessionized_metrics::FailureReasons
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
         }
     }
 }
diff --git a/crates/analytics/src/payments/metrics/avg_ticket_size.rs b/crates/analytics/src/payments/metrics/avg_ticket_size.rs
index ec76bfc86dc..fc2f44cada2 100644
--- a/crates/analytics/src/payments/metrics/avg_ticket_size.rs
+++ b/crates/analytics/src/payments/metrics/avg_ticket_size.rs
@@ -117,6 +117,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/metrics/connector_success_rate.rs b/crates/analytics/src/payments/metrics/connector_success_rate.rs
index e44e87a7e5f..36783eda72a 100644
--- a/crates/analytics/src/payments/metrics/connector_success_rate.rs
+++ b/crates/analytics/src/payments/metrics/connector_success_rate.rs
@@ -112,6 +112,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/metrics/payment_count.rs b/crates/analytics/src/payments/metrics/payment_count.rs
index 642ba9ef035..bd0b52d7cf8 100644
--- a/crates/analytics/src/payments/metrics/payment_count.rs
+++ b/crates/analytics/src/payments/metrics/payment_count.rs
@@ -103,6 +103,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/metrics/payment_processed_amount.rs b/crates/analytics/src/payments/metrics/payment_processed_amount.rs
index c31b8873cb8..b8b3868803c 100644
--- a/crates/analytics/src/payments/metrics/payment_processed_amount.rs
+++ b/crates/analytics/src/payments/metrics/payment_processed_amount.rs
@@ -111,6 +111,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/metrics/payment_success_count.rs b/crates/analytics/src/payments/metrics/payment_success_count.rs
index a45a4557c52..ea926761c13 100644
--- a/crates/analytics/src/payments/metrics/payment_success_count.rs
+++ b/crates/analytics/src/payments/metrics/payment_success_count.rs
@@ -110,6 +110,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/metrics/retries_count.rs b/crates/analytics/src/payments/metrics/retries_count.rs
index c3c7d40e321..9695e1fe18b 100644
--- a/crates/analytics/src/payments/metrics/retries_count.rs
+++ b/crates/analytics/src/payments/metrics/retries_count.rs
@@ -107,6 +107,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/avg_ticket_size.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/avg_ticket_size.rs
new file mode 100644
index 00000000000..b7f13667917
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/avg_ticket_size.rs
@@ -0,0 +1,146 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::{PaymentMetric, PaymentMetricRow};
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct AvgTicketSize;
+
+#[async_trait::async_trait]
+impl<T> PaymentMetric<T> for AvgTicketSize
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Sum {
+                field: "amount",
+                alias: Some("total"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .add_filter_clause(
+                PaymentDimensions::PaymentStatus,
+                storage_enums::AttemptStatus::Charged,
+            )
+            .switch()?;
+
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        i.status.as_ref().map(|i| i.0),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/connector_success_rate.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/connector_success_rate.rs
new file mode 100644
index 00000000000..66006c15a2a
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/connector_success_rate.rs
@@ -0,0 +1,141 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct ConnectorSuccessRate;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for ConnectorSuccessRate
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentDimensions::PaymentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        query_builder
+            .add_custom_filter_clause(PaymentDimensions::Connector, "NULL", FilterTypes::IsNotNull)
+            .switch()?;
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/failure_reasons.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/failure_reasons.rs
new file mode 100644
index 00000000000..70ae64e0115
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/failure_reasons.rs
@@ -0,0 +1,208 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{
+        Aggregate, FilterTypes, GroupByClause, Order, QueryBuilder, QueryFilter, SeriesBucket,
+        ToSql, Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct FailureReasons;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for FailureReasons
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut inner_query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+        inner_query_builder
+            .add_select_column("sum(sign_flag)")
+            .switch()?;
+
+        inner_query_builder
+            .add_custom_filter_clause(
+                PaymentDimensions::ErrorReason,
+                "NULL",
+                FilterTypes::IsNotNull,
+            )
+            .switch()?;
+
+        time_range
+            .set_filter_clause(&mut inner_query_builder)
+            .attach_printable("Error filtering time range for inner query")
+            .switch()?;
+
+        let inner_query_string = inner_query_builder
+            .build_query()
+            .attach_printable("Error building inner query")
+            .change_context(MetricsError::QueryBuildingError)?;
+
+        let mut outer_query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+
+        for dim in dimensions.iter() {
+            outer_query_builder.add_select_column(dim).switch()?;
+        }
+
+        outer_query_builder
+            .add_select_column("sum(sign_flag) AS count")
+            .switch()?;
+
+        outer_query_builder
+            .add_select_column(format!("({}) AS total", inner_query_string))
+            .switch()?;
+
+        outer_query_builder
+            .add_select_column("first_attempt")
+            .switch()?;
+
+        outer_query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+
+        outer_query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters
+            .set_filter_clause(&mut outer_query_builder)
+            .switch()?;
+
+        auth.set_filter_clause(&mut outer_query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut outer_query_builder)
+            .attach_printable("Error filtering time range for outer query")
+            .switch()?;
+
+        outer_query_builder
+            .add_filter_clause(
+                PaymentDimensions::PaymentStatus,
+                storage_enums::AttemptStatus::Failure,
+            )
+            .switch()?;
+
+        outer_query_builder
+            .add_custom_filter_clause(
+                PaymentDimensions::ErrorReason,
+                "NULL",
+                FilterTypes::IsNotNull,
+            )
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            outer_query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        outer_query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut outer_query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        outer_query_builder
+            .add_order_by_clause("count", Order::Descending)
+            .attach_printable("Error adding order by clause")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            if dim != &PaymentDimensions::ErrorReason {
+                outer_query_builder
+                    .add_order_by_clause(dim, Order::Ascending)
+                    .attach_printable("Error adding order by clause")
+                    .switch()?;
+            }
+        }
+
+        outer_query_builder
+            .set_limit_by(5, &[PaymentDimensions::Connector])
+            .attach_printable("Error adding limit clause")
+            .switch()?;
+
+        outer_query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/mod.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/mod.rs
new file mode 100644
index 00000000000..e3a5e370534
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/mod.rs
@@ -0,0 +1,20 @@
+mod avg_ticket_size;
+mod connector_success_rate;
+mod failure_reasons;
+mod payment_count;
+mod payment_processed_amount;
+mod payment_success_count;
+mod payments_distribution;
+mod retries_count;
+mod success_rate;
+pub(super) use avg_ticket_size::AvgTicketSize;
+pub(super) use connector_success_rate::ConnectorSuccessRate;
+pub(super) use failure_reasons::FailureReasons;
+pub(super) use payment_count::PaymentCount;
+pub(super) use payment_processed_amount::PaymentProcessedAmount;
+pub(super) use payment_success_count::PaymentSuccessCount;
+pub(super) use payments_distribution::PaymentsDistribution;
+pub(super) use retries_count::RetriesCount;
+pub(super) use success_rate::PaymentSuccessRate;
+
+pub use super::{PaymentMetric, PaymentMetricAnalytics, PaymentMetricRow};
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/payment_count.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/payment_count.rs
new file mode 100644
index 00000000000..29d083dd232
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/payment_count.rs
@@ -0,0 +1,129 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentCount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for PaymentCount
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        i.status.as_ref().map(|i| i.0),
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<HashSet<_>, crate::query::PostProcessingError>>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/payment_processed_amount.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/payment_processed_amount.rs
new file mode 100644
index 00000000000..9bc554eaae7
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/payment_processed_amount.rs
@@ -0,0 +1,155 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentProcessedAmount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for PaymentProcessedAmount
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentDimensions::PaymentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder.add_select_column("first_attempt").switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Sum {
+                field: "amount",
+                alias: Some("total"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+        query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .add_filter_clause(
+                PaymentDimensions::PaymentStatus,
+                storage_enums::AttemptStatus::Charged,
+            )
+            .switch()?;
+
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/payment_success_count.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/payment_success_count.rs
new file mode 100644
index 00000000000..b307b77d5da
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/payment_success_count.rs
@@ -0,0 +1,139 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentSuccessCount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for PaymentSuccessCount
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .add_filter_clause(
+                PaymentDimensions::PaymentStatus,
+                storage_enums::AttemptStatus::Charged,
+            )
+            .switch()?;
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/payments_distribution.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/payments_distribution.rs
new file mode 100644
index 00000000000..e0987dd0d22
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/payments_distribution.rs
@@ -0,0 +1,142 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentsDistribution;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for PaymentsDistribution
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentDimensions::PaymentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder.add_select_column("first_attempt").switch()?;
+
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+        query_builder
+            .add_group_by_clause("first_attempt")
+            .attach_printable("Error grouping by first_attempt")
+            .switch()?;
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/retries_count.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/retries_count.rs
new file mode 100644
index 00000000000..b79489744cc
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/retries_count.rs
@@ -0,0 +1,135 @@
+use std::collections::HashSet;
+
+use api_models::{
+    analytics::{
+        payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+        Granularity, TimeRange,
+    },
+    enums::IntentStatus,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct RetriesCount;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for RetriesCount
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        _dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        _filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntentSessionized);
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Sum {
+                field: "amount",
+                alias: Some("total"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        query_builder
+            .add_custom_filter_clause("attempt_count", "1", FilterTypes::Gt)
+            .switch()?;
+        query_builder
+            .add_custom_filter_clause("status", IntentStatus::Succeeded, FilterTypes::Equal)
+            .switch()?;
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/sessionized_metrics/success_rate.rs b/crates/analytics/src/payments/metrics/sessionized_metrics/success_rate.rs
new file mode 100644
index 00000000000..30e7471d608
--- /dev/null
+++ b/crates/analytics/src/payments/metrics/sessionized_metrics/success_rate.rs
@@ -0,0 +1,135 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::PaymentMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct PaymentSuccessRate;
+
+#[async_trait::async_trait]
+impl<T> super::PaymentMetric<T> for PaymentSuccessRate
+where
+    T: AnalyticsDataSource + super::PaymentMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentDimensions],
+        auth: &AuthInfo,
+        filters: &PaymentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentSessionized);
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(PaymentDimensions::PaymentStatus);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Count {
+                field: None,
+                alias: Some("count"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Min {
+                field: "created_at",
+                alias: Some("start_bucket"),
+            })
+            .switch()?;
+        query_builder
+            .add_select_column(Aggregate::Max {
+                field: "created_at",
+                alias: Some("end_bucket"),
+            })
+            .switch()?;
+
+        filters.set_filter_clause(&mut query_builder).switch()?;
+
+        auth.set_filter_clause(&mut query_builder).switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder
+                .add_group_by_clause(dim)
+                .attach_printable("Error grouping by dimensions")
+                .switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .attach_printable("Error adding granularity")
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<PaymentMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    PaymentMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.authentication_type.as_ref().map(|i| i.0),
+                        i.payment_method.clone(),
+                        i.payment_method_type.clone(),
+                        i.client_source.clone(),
+                        i.client_version.clone(),
+                        i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
+                        TimeRange {
+                            start_time: match (granularity, i.start_bucket) {
+                                (Some(g), Some(st)) => g.clip_to_start(st)?,
+                                _ => time_range.start_time,
+                            },
+                            end_time: granularity.as_ref().map_or_else(
+                                || Ok(time_range.end_time),
+                                |g| i.end_bucket.map(|et| g.clip_to_end(et)).transpose(),
+                            )?,
+                        },
+                    ),
+                    i,
+                ))
+            })
+            .collect::<error_stack::Result<
+                HashSet<(PaymentMetricsBucketIdentifier, PaymentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/success_rate.rs b/crates/analytics/src/payments/metrics/success_rate.rs
index b8e840c2203..e756115a67d 100644
--- a/crates/analytics/src/payments/metrics/success_rate.rs
+++ b/crates/analytics/src/payments/metrics/success_rate.rs
@@ -106,6 +106,11 @@ where
                         i.client_source.clone(),
                         i.client_version.clone(),
                         i.profile_id.clone(),
+                        i.card_network.clone(),
+                        i.merchant_id.clone(),
+                        i.card_last_4.clone(),
+                        i.card_issuer.clone(),
+                        i.error_reason.clone(),
                         TimeRange {
                             start_time: match (granularity, i.start_bucket) {
                                 (Some(g), Some(st)) => g.clip_to_start(st)?,
diff --git a/crates/analytics/src/payments/types.rs b/crates/analytics/src/payments/types.rs
index b7984a19ea3..e23fc6cbbb3 100644
--- a/crates/analytics/src/payments/types.rs
+++ b/crates/analytics/src/payments/types.rs
@@ -48,7 +48,7 @@ where
                     PaymentDimensions::PaymentMethodType,
                     &self.payment_method_type,
                 )
-                .attach_printable("Error adding payment method filter")?;
+                .attach_printable("Error adding payment method type filter")?;
         }
         if !self.client_source.is_empty() {
             builder
@@ -84,6 +84,26 @@ where
                 )
                 .attach_printable("Error adding card network filter")?;
         }
+        if !self.merchant_id.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentDimensions::MerchantId, &self.merchant_id)
+                .attach_printable("Error adding merchant id filter")?;
+        }
+        if !self.card_last_4.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentDimensions::CardLast4, &self.card_last_4)
+                .attach_printable("Error adding card last 4 filter")?;
+        }
+        if !self.card_issuer.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentDimensions::CardIssuer, &self.card_issuer)
+                .attach_printable("Error adding card issuer filter")?;
+        }
+        if !self.error_reason.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentDimensions::ErrorReason, &self.error_reason)
+                .attach_printable("Error adding error reason filter")?;
+        }
         Ok(())
     }
 }
diff --git a/crates/analytics/src/query.rs b/crates/analytics/src/query.rs
index 5f8a2f4a330..7ce338f7db3 100644
--- a/crates/analytics/src/query.rs
+++ b/crates/analytics/src/query.rs
@@ -1,4 +1,4 @@
-use std::marker::PhantomData;
+use std::{fmt, marker::PhantomData};
 
 use api_models::{
     analytics::{
@@ -301,8 +301,8 @@ pub enum Order {
     Descending,
 }
 
-impl std::fmt::Display for Order {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+impl fmt::Display for Order {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::Ascending => write!(f, "asc"),
             Self::Descending => write!(f, "desc"),
@@ -335,6 +335,68 @@ pub struct TopN {
     pub order: Order,
 }
 
+#[derive(Debug, Clone)]
+pub struct LimitByClause {
+    limit: u64,
+    columns: Vec<String>,
+}
+
+impl fmt::Display for LimitByClause {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "LIMIT {} BY {}", self.limit, self.columns.join(", "))
+    }
+}
+
+#[derive(Debug, Default, Clone, Copy)]
+pub enum FilterCombinator {
+    #[default]
+    And,
+    Or,
+}
+
+impl<T: AnalyticsDataSource> ToSql<T> for FilterCombinator {
+    fn to_sql(&self, _table_engine: &TableEngine) -> error_stack::Result<String, ParsingError> {
+        Ok(match self {
+            Self::And => " AND ",
+            Self::Or => " OR ",
+        }
+        .to_owned())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum Filter {
+    Plain(String, FilterTypes, String),
+    NestedFilter(FilterCombinator, Vec<Filter>),
+}
+
+impl Default for Filter {
+    fn default() -> Self {
+        Self::NestedFilter(FilterCombinator::default(), Vec::new())
+    }
+}
+
+impl<T: AnalyticsDataSource> ToSql<T> for Filter {
+    fn to_sql(&self, table_engine: &TableEngine) -> error_stack::Result<String, ParsingError> {
+        Ok(match self {
+            Self::Plain(l, op, r) => filter_type_to_sql(l, op, r),
+            Self::NestedFilter(operator, filters) => {
+                format!(
+                    "( {} )",
+                    filters
+                        .iter()
+                        .map(|f| <Self as ToSql<T>>::to_sql(f, table_engine))
+                        .collect::<Result<Vec<String>, _>>()?
+                        .join(
+                            <FilterCombinator as ToSql<T>>::to_sql(operator, table_engine)?
+                                .as_ref()
+                        )
+                )
+            }
+        })
+    }
+}
+
 #[derive(Debug)]
 pub struct QueryBuilder<T>
 where
@@ -342,9 +404,11 @@ where
     AnalyticsCollection: ToSql<T>,
 {
     columns: Vec<String>,
-    filters: Vec<(String, FilterTypes, String)>,
+    filters: Filter,
     group_by: Vec<String>,
+    order_by: Vec<String>,
     having: Option<Vec<(String, FilterTypes, String)>>,
+    limit_by: Option<LimitByClause>,
     outer_select: Vec<String>,
     top_n: Option<TopN>,
     table: AnalyticsCollection,
@@ -444,7 +508,7 @@ impl_to_sql_for_to_string!(
     DisputeStage
 );
 
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy)]
 pub enum FilterTypes {
     Equal,
     NotEqual,
@@ -483,7 +547,9 @@ where
             columns: Default::default(),
             filters: Default::default(),
             group_by: Default::default(),
+            order_by: Default::default(),
             having: Default::default(),
+            limit_by: Default::default(),
             outer_select: Default::default(),
             top_n: Default::default(),
             table,
@@ -580,7 +646,7 @@ where
         rhs: impl ToSql<T>,
         comparison: FilterTypes,
     ) -> QueryResult<()> {
-        self.filters.push((
+        let filter = Filter::Plain(
             lhs.to_sql(&self.table_engine)
                 .change_context(QueryBuildingError::SqlSerializeError)
                 .attach_printable("Error serializing filter key")?,
@@ -588,9 +654,18 @@ where
             rhs.to_sql(&self.table_engine)
                 .change_context(QueryBuildingError::SqlSerializeError)
                 .attach_printable("Error serializing filter value")?,
-        ));
+        );
+        self.add_nested_filter_clause(filter);
         Ok(())
     }
+    pub fn add_nested_filter_clause(&mut self, filter: Filter) {
+        match &mut self.filters {
+            Filter::NestedFilter(_, ref mut filters) => filters.push(filter),
+            f @ Filter::Plain(_, _, _) => {
+                self.filters = Filter::NestedFilter(FilterCombinator::And, vec![f.clone(), filter]);
+            }
+        }
+    }
 
     pub fn add_filter_in_range_clause(
         &mut self,
@@ -623,6 +698,37 @@ where
         Ok(())
     }
 
+    pub fn add_order_by_clause(
+        &mut self,
+        column: impl ToSql<T>,
+        order: impl ToSql<T>,
+    ) -> QueryResult<()> {
+        let column_sql = column
+            .to_sql(&self.table_engine)
+            .change_context(QueryBuildingError::SqlSerializeError)
+            .attach_printable("Error serializing order by column")?;
+
+        let order_sql = order
+            .to_sql(&self.table_engine)
+            .change_context(QueryBuildingError::SqlSerializeError)
+            .attach_printable("Error serializing order direction")?;
+
+        self.order_by.push(format!("{} {}", column_sql, order_sql));
+        Ok(())
+    }
+
+    pub fn set_limit_by(&mut self, limit: u64, columns: &[impl ToSql<T>]) -> QueryResult<()> {
+        let columns = columns
+            .iter()
+            .map(|col| col.to_sql(&self.table_engine))
+            .collect::<Result<Vec<String>, _>>()
+            .change_context(QueryBuildingError::SqlSerializeError)
+            .attach_printable("Error serializing LIMIT BY columns")?;
+
+        self.limit_by = Some(LimitByClause { limit, columns });
+        Ok(())
+    }
+
     pub fn add_granularity_in_mins(&mut self, granularity: &Granularity) -> QueryResult<()> {
         let interval = match granularity {
             Granularity::OneMin => "1",
@@ -638,12 +744,9 @@ where
         Ok(())
     }
 
-    fn get_filter_clause(&self) -> String {
-        self.filters
-            .iter()
-            .map(|(l, op, r)| filter_type_to_sql(l, op, r))
-            .collect::<Vec<String>>()
-            .join(" AND ")
+    fn get_filter_clause(&self) -> QueryResult<String> {
+        <Filter as ToSql<T>>::to_sql(&self.filters, &self.table_engine)
+            .change_context(QueryBuildingError::SqlSerializeError)
     }
 
     fn get_select_clause(&self) -> String {
@@ -731,9 +834,10 @@ where
                 .attach_printable("Error serializing table value")?,
         );
 
-        if !self.filters.is_empty() {
+        let filter_clause = self.get_filter_clause()?;
+        if !filter_clause.is_empty() {
             query.push_str(" WHERE ");
-            query.push_str(&self.get_filter_clause());
+            query.push_str(filter_clause.as_str());
         }
 
         if !self.group_by.is_empty() {
@@ -758,6 +862,15 @@ where
             }
         }
 
+        if !self.order_by.is_empty() {
+            query.push_str(" ORDER BY ");
+            query.push_str(&self.order_by.join(", "));
+        }
+
+        if let Some(limit_by) = &self.limit_by {
+            query.push_str(&format!(" {}", limit_by));
+        }
+
         if !self.outer_select.is_empty() {
             query.insert_str(
                 0,
diff --git a/crates/analytics/src/sqlx.rs b/crates/analytics/src/sqlx.rs
index e9395439209..ae72bbeffea 100644
--- a/crates/analytics/src/sqlx.rs
+++ b/crates/analytics/src/sqlx.rs
@@ -330,6 +330,30 @@ impl<'a> FromRow<'a, PgRow> for super::payments::metrics::PaymentMetricRow {
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
+        let card_network: Option<String> = row.try_get("card_network").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let merchant_id: Option<String> = row.try_get("merchant_id").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_last_4: Option<String> = row.try_get("card_last_4").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_issuer: Option<String> = row.try_get("card_issuer").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let error_reason: Option<String> = row.try_get("error_reason").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let first_attempt: Option<bool> = row.try_get("first_attempt").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         let total: Option<bigdecimal::BigDecimal> = row.try_get("total").or_else(|e| match e {
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
@@ -355,6 +379,12 @@ impl<'a> FromRow<'a, PgRow> for super::payments::metrics::PaymentMetricRow {
             client_source,
             client_version,
             profile_id,
+            card_network,
+            merchant_id,
+            card_last_4,
+            card_issuer,
+            error_reason,
+            first_attempt,
             total,
             count,
             start_bucket,
@@ -407,6 +437,26 @@ impl<'a> FromRow<'a, PgRow> for super::payments::distribution::PaymentDistributi
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
+        let card_network: Option<String> = row.try_get("card_network").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let merchant_id: Option<String> = row.try_get("merchant_id").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_last_4: Option<String> = row.try_get("card_last_4").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_issuer: Option<String> = row.try_get("card_issuer").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let error_reason: Option<String> = row.try_get("error_reason").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         let total: Option<bigdecimal::BigDecimal> = row.try_get("total").or_else(|e| match e {
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
@@ -419,6 +469,10 @@ impl<'a> FromRow<'a, PgRow> for super::payments::distribution::PaymentDistributi
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
+        let first_attempt: Option<bool> = row.try_get("first_attempt").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         // Removing millisecond precision to get accurate diffs against clickhouse
         let start_bucket: Option<PrimitiveDateTime> = row
             .try_get::<Option<PrimitiveDateTime>, _>("start_bucket")?
@@ -436,6 +490,12 @@ impl<'a> FromRow<'a, PgRow> for super::payments::distribution::PaymentDistributi
             client_source,
             client_version,
             profile_id,
+            card_network,
+            merchant_id,
+            card_last_4,
+            card_issuer,
+            error_reason,
+            first_attempt,
             total,
             count,
             error_message,
@@ -493,6 +553,22 @@ impl<'a> FromRow<'a, PgRow> for super::payments::filters::PaymentFilterRow {
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
+        let merchant_id: Option<String> = row.try_get("merchant_id").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_last_4: Option<String> = row.try_get("card_last_4").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_issuer: Option<String> = row.try_get("card_issuer").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let error_reason: Option<String> = row.try_get("error_reason").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         Ok(Self {
             currency,
             status,
@@ -504,6 +580,10 @@ impl<'a> FromRow<'a, PgRow> for super::payments::filters::PaymentFilterRow {
             client_version,
             profile_id,
             card_network,
+            merchant_id,
+            card_last_4,
+            card_issuer,
+            error_reason,
         })
     }
 }
@@ -524,6 +604,45 @@ impl<'a> FromRow<'a, PgRow> for super::payment_intents::metrics::PaymentIntentMe
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
+        let connector: Option<String> = row.try_get("connector").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let authentication_type: Option<DBEnumWrapper<AuthenticationType>> =
+            row.try_get("authentication_type").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let payment_method: Option<String> =
+            row.try_get("payment_method").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let payment_method_type: Option<String> =
+            row.try_get("payment_method_type").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let card_network: Option<String> = row.try_get("card_network").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let merchant_id: Option<String> = row.try_get("merchant_id").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_last_4: Option<String> = row.try_get("card_last_4").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_issuer: Option<String> = row.try_get("card_issuer").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let error_reason: Option<String> = row.try_get("error_reason").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         let total: Option<bigdecimal::BigDecimal> = row.try_get("total").or_else(|e| match e {
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
@@ -532,6 +651,10 @@ impl<'a> FromRow<'a, PgRow> for super::payment_intents::metrics::PaymentIntentMe
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
+        let first_attempt: Option<i64> = row.try_get("first_attempt").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         // Removing millisecond precision to get accurate diffs against clickhouse
         let start_bucket: Option<PrimitiveDateTime> = row
             .try_get::<Option<PrimitiveDateTime>, _>("start_bucket")?
@@ -543,6 +666,16 @@ impl<'a> FromRow<'a, PgRow> for super::payment_intents::metrics::PaymentIntentMe
             status,
             currency,
             profile_id,
+            connector,
+            authentication_type,
+            payment_method,
+            payment_method_type,
+            card_network,
+            merchant_id,
+            card_last_4,
+            card_issuer,
+            error_reason,
+            first_attempt,
             total,
             count,
             start_bucket,
@@ -567,11 +700,58 @@ impl<'a> FromRow<'a, PgRow> for super::payment_intents::filters::PaymentIntentFi
             ColumnNotFound(_) => Ok(Default::default()),
             e => Err(e),
         })?;
-
+        let connector: Option<String> = row.try_get("connector").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let authentication_type: Option<DBEnumWrapper<AuthenticationType>> =
+            row.try_get("authentication_type").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let payment_method: Option<String> =
+            row.try_get("payment_method").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let payment_method_type: Option<String> =
+            row.try_get("payment_method_type").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let card_network: Option<String> = row.try_get("card_network").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let merchant_id: Option<String> = row.try_get("merchant_id").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_last_4: Option<String> = row.try_get("card_last_4").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let card_issuer: Option<String> = row.try_get("card_issuer").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
+        let error_reason: Option<String> = row.try_get("error_reason").or_else(|e| match e {
+            ColumnNotFound(_) => Ok(Default::default()),
+            e => Err(e),
+        })?;
         Ok(Self {
             status,
             currency,
             profile_id,
+            connector,
+            authentication_type,
+            payment_method,
+            payment_method_type,
+            card_network,
+            merchant_id,
+            card_last_4,
+            card_issuer,
+            error_reason,
         })
     }
 }
@@ -716,7 +896,11 @@ impl ToSql<SqlxClient> for AnalyticsCollection {
     fn to_sql(&self, _table_engine: &TableEngine) -> error_stack::Result<String, ParsingError> {
         match self {
             Self::Payment => Ok("payment_attempt".to_string()),
+            Self::PaymentSessionized => Err(error_stack::report!(ParsingError::UnknownError)
+                .attach_printable("PaymentSessionized table is not implemented for Sqlx"))?,
             Self::Refund => Ok("refund".to_string()),
+            Self::RefundSessionized => Err(error_stack::report!(ParsingError::UnknownError)
+                .attach_printable("RefundSessionized table is not implemented for Sqlx"))?,
             Self::SdkEvents => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("SdkEventsAudit table is not implemented for Sqlx"))?,
             Self::SdkEventsAnalytics => Err(error_stack::report!(ParsingError::UnknownError)
@@ -725,6 +909,10 @@ impl ToSql<SqlxClient> for AnalyticsCollection {
                 .attach_printable("ApiEvents table is not implemented for Sqlx"))?,
             Self::FraudCheck => Ok("fraud_check".to_string()),
             Self::PaymentIntent => Ok("payment_intent".to_string()),
+            Self::PaymentIntentSessionized => Err(error_stack::report!(
+                ParsingError::UnknownError
+            )
+            .attach_printable("PaymentIntentSessionized table is not implemented for Sqlx"))?,
             Self::ConnectorEvents => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("ConnectorEvents table is not implemented for Sqlx"))?,
             Self::ApiEventsAnalytics => Err(error_stack::report!(ParsingError::UnknownError)
diff --git a/crates/analytics/src/types.rs b/crates/analytics/src/types.rs
index 6593e5d6eb0..6bdd11fcd73 100644
--- a/crates/analytics/src/types.rs
+++ b/crates/analytics/src/types.rs
@@ -26,12 +26,15 @@ pub enum AnalyticsDomain {
 #[derive(Debug, strum::AsRefStr, strum::Display, Clone, Copy)]
 pub enum AnalyticsCollection {
     Payment,
+    PaymentSessionized,
     Refund,
+    RefundSessionized,
     FraudCheck,
     SdkEvents,
     SdkEventsAnalytics,
     ApiEvents,
     PaymentIntent,
+    PaymentIntentSessionized,
     ConnectorEvents,
     OutgoingWebhookEvent,
     Dispute,
@@ -56,6 +59,12 @@ impl<T: FromStr + Display> AsRef<T> for DBEnumWrapper<T> {
     }
 }
 
+impl<T: FromStr + Display + Default> Default for DBEnumWrapper<T> {
+    fn default() -> Self {
+        Self(T::default())
+    }
+}
+
 impl<T> FromStr for DBEnumWrapper<T>
 where
     T: FromStr + Display,
diff --git a/crates/analytics/src/utils.rs b/crates/analytics/src/utils.rs
index 7b73f5a1c1d..fc21bf09819 100644
--- a/crates/analytics/src/utils.rs
+++ b/crates/analytics/src/utils.rs
@@ -12,11 +12,39 @@ use api_models::analytics::{
 use strum::IntoEnumIterator;
 
 pub fn get_payment_dimensions() -> Vec<NameDescription> {
-    PaymentDimensions::iter().map(Into::into).collect()
+    vec![
+        PaymentDimensions::Connector,
+        PaymentDimensions::PaymentMethod,
+        PaymentDimensions::PaymentMethodType,
+        PaymentDimensions::Currency,
+        PaymentDimensions::AuthType,
+        PaymentDimensions::PaymentStatus,
+        PaymentDimensions::ClientSource,
+        PaymentDimensions::ClientVersion,
+        PaymentDimensions::ProfileId,
+        PaymentDimensions::CardNetwork,
+        PaymentDimensions::MerchantId,
+    ]
+    .into_iter()
+    .map(Into::into)
+    .collect()
 }
 
 pub fn get_payment_intent_dimensions() -> Vec<NameDescription> {
-    PaymentIntentDimensions::iter().map(Into::into).collect()
+    vec![
+        PaymentIntentDimensions::PaymentIntentStatus,
+        PaymentIntentDimensions::Currency,
+        PaymentIntentDimensions::ProfileId,
+        PaymentIntentDimensions::Connector,
+        PaymentIntentDimensions::AuthType,
+        PaymentIntentDimensions::PaymentMethod,
+        PaymentIntentDimensions::PaymentMethodType,
+        PaymentIntentDimensions::CardNetwork,
+        PaymentIntentDimensions::MerchantId,
+    ]
+    .into_iter()
+    .map(Into::into)
+    .collect()
 }
 
 pub fn get_refund_dimensions() -> Vec<NameDescription> {
diff --git a/crates/api_models/src/analytics.rs b/crates/api_models/src/analytics.rs
index 0379ec09547..b95404080b0 100644
--- a/crates/api_models/src/analytics.rs
+++ b/crates/api_models/src/analytics.rs
@@ -200,6 +200,28 @@ pub struct AnalyticsMetadata {
     pub current_time_range: TimeRange,
 }
 
+#[derive(Debug, serde::Serialize)]
+pub struct PaymentsAnalyticsMetadata {
+    pub total_payment_processed_amount: Option<u64>,
+    pub total_payment_processed_amount_without_smart_retries: Option<u64>,
+    pub total_payment_processed_count: Option<u64>,
+    pub total_payment_processed_count_without_smart_retries: Option<u64>,
+    pub total_failure_reasons_count: Option<u64>,
+    pub total_failure_reasons_count_without_smart_retries: Option<u64>,
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct PaymentIntentsAnalyticsMetadata {
+    pub total_success_rate: Option<f64>,
+    pub total_success_rate_without_smart_retries: Option<f64>,
+    pub total_smart_retried_amount: Option<u64>,
+    pub total_smart_retried_amount_without_smart_retries: Option<u64>,
+    pub total_payment_processed_amount: Option<u64>,
+    pub total_payment_processed_amount_without_smart_retries: Option<u64>,
+    pub total_payment_processed_count: Option<u64>,
+    pub total_payment_processed_count_without_smart_retries: Option<u64>,
+}
+
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GetPaymentFiltersRequest {
@@ -320,6 +342,20 @@ pub struct MetricsResponse<T> {
     pub meta_data: [AnalyticsMetadata; 1],
 }
 
+#[derive(Debug, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PaymentsMetricsResponse<T> {
+    pub query_data: Vec<T>,
+    pub meta_data: [PaymentsAnalyticsMetadata; 1],
+}
+
+#[derive(Debug, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PaymentIntentsMetricsResponse<T> {
+    pub query_data: Vec<T>,
+    pub meta_data: [PaymentIntentsAnalyticsMetadata; 1],
+}
+
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GetApiEventFiltersRequest {
@@ -390,3 +426,21 @@ pub struct GetDisputeMetricRequest {
     #[serde(default)]
     pub delta: bool,
 }
+
+#[derive(Clone, Debug, Default, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+pub struct SankeyResponse {
+    pub normal_success: i64,
+    pub normal_failure: i64,
+    pub cancelled: i64,
+    pub smart_retried_success: i64,
+    pub smart_retried_failure: i64,
+    pub pending: i64,
+    pub partial_refunded: i64,
+    pub refunded: i64,
+    pub disputed: i64,
+    pub pm_awaited: i64,
+    pub customer_awaited: i64,
+    pub merchant_awaited: i64,
+    pub confirmation_awaited: i64,
+}
diff --git a/crates/api_models/src/analytics/payment_intents.rs b/crates/api_models/src/analytics/payment_intents.rs
index 61e336185b6..41f11c19ef8 100644
--- a/crates/api_models/src/analytics/payment_intents.rs
+++ b/crates/api_models/src/analytics/payment_intents.rs
@@ -6,7 +6,9 @@ use std::{
 use common_utils::id_type;
 
 use super::{NameDescription, TimeRange};
-use crate::enums::{Currency, IntentStatus};
+use crate::enums::{
+    AuthenticationType, Connector, Currency, IntentStatus, PaymentMethod, PaymentMethodType,
+};
 
 #[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct PaymentIntentFilters {
@@ -16,6 +18,24 @@ pub struct PaymentIntentFilters {
     pub currency: Vec<Currency>,
     #[serde(default)]
     pub profile_id: Vec<id_type::ProfileId>,
+    #[serde(default)]
+    pub connector: Vec<Connector>,
+    #[serde(default)]
+    pub auth_type: Vec<AuthenticationType>,
+    #[serde(default)]
+    pub payment_method: Vec<PaymentMethod>,
+    #[serde(default)]
+    pub payment_method_type: Vec<PaymentMethodType>,
+    #[serde(default)]
+    pub card_network: Vec<String>,
+    #[serde(default)]
+    pub merchant_id: Vec<id_type::MerchantId>,
+    #[serde(default)]
+    pub card_last_4: Vec<String>,
+    #[serde(default)]
+    pub card_issuer: Vec<String>,
+    #[serde(default)]
+    pub error_reason: Vec<String>,
 }
 
 #[derive(
@@ -40,6 +60,15 @@ pub enum PaymentIntentDimensions {
     PaymentIntentStatus,
     Currency,
     ProfileId,
+    Connector,
+    AuthType,
+    PaymentMethod,
+    PaymentMethodType,
+    CardNetwork,
+    MerchantId,
+    CardLast4,
+    CardIssuer,
+    ErrorReason,
 }
 
 #[derive(
@@ -61,6 +90,14 @@ pub enum PaymentIntentMetrics {
     TotalSmartRetries,
     SmartRetriedAmount,
     PaymentIntentCount,
+    PaymentsSuccessRate,
+    SessionizedSuccessfulSmartRetries,
+    SessionizedTotalSmartRetries,
+    SessionizedSmartRetriedAmount,
+    SessionizedPaymentIntentCount,
+    SessionizedPaymentsSuccessRate,
+    SessionizedPaymentProcessedAmount,
+    SessionizedPaymentsDistribution,
 }
 
 #[derive(Debug, Default, serde::Serialize)]
@@ -75,6 +112,7 @@ pub mod metric_behaviour {
     pub struct TotalSmartRetries;
     pub struct SmartRetriedAmount;
     pub struct PaymentIntentCount;
+    pub struct PaymentsSuccessRate;
 }
 
 impl From<PaymentIntentMetrics> for NameDescription {
@@ -100,6 +138,15 @@ pub struct PaymentIntentMetricsBucketIdentifier {
     pub status: Option<IntentStatus>,
     pub currency: Option<Currency>,
     pub profile_id: Option<String>,
+    pub connector: Option<String>,
+    pub auth_type: Option<AuthenticationType>,
+    pub payment_method: Option<String>,
+    pub payment_method_type: Option<String>,
+    pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
     #[serde(rename = "time_range")]
     pub time_bucket: TimeRange,
     #[serde(rename = "time_bucket")]
@@ -113,12 +160,30 @@ impl PaymentIntentMetricsBucketIdentifier {
         status: Option<IntentStatus>,
         currency: Option<Currency>,
         profile_id: Option<String>,
+        connector: Option<String>,
+        auth_type: Option<AuthenticationType>,
+        payment_method: Option<String>,
+        payment_method_type: Option<String>,
+        card_network: Option<String>,
+        merchant_id: Option<String>,
+        card_last_4: Option<String>,
+        card_issuer: Option<String>,
+        error_reason: Option<String>,
         normalized_time_range: TimeRange,
     ) -> Self {
         Self {
             status,
             currency,
             profile_id,
+            connector,
+            auth_type,
+            payment_method,
+            payment_method_type,
+            card_network,
+            merchant_id,
+            card_last_4,
+            card_issuer,
+            error_reason,
             time_bucket: normalized_time_range,
             start_time: normalized_time_range.start_time,
         }
@@ -130,6 +195,15 @@ impl Hash for PaymentIntentMetricsBucketIdentifier {
         self.status.map(|i| i.to_string()).hash(state);
         self.currency.hash(state);
         self.profile_id.hash(state);
+        self.connector.hash(state);
+        self.auth_type.map(|i| i.to_string()).hash(state);
+        self.payment_method.hash(state);
+        self.payment_method_type.hash(state);
+        self.card_network.hash(state);
+        self.merchant_id.hash(state);
+        self.card_last_4.hash(state);
+        self.card_issuer.hash(state);
+        self.error_reason.hash(state);
         self.time_bucket.hash(state);
     }
 }
@@ -149,7 +223,19 @@ pub struct PaymentIntentMetricsBucketValue {
     pub successful_smart_retries: Option<u64>,
     pub total_smart_retries: Option<u64>,
     pub smart_retried_amount: Option<u64>,
+    pub smart_retried_amount_without_smart_retries: Option<u64>,
     pub payment_intent_count: Option<u64>,
+    pub successful_payments: Option<u32>,
+    pub successful_payments_without_smart_retries: Option<u32>,
+    pub total_payments: Option<u32>,
+    pub payments_success_rate: Option<f64>,
+    pub payments_success_rate_without_smart_retries: Option<f64>,
+    pub payment_processed_amount: Option<u64>,
+    pub payment_processed_count: Option<u64>,
+    pub payment_processed_amount_without_smart_retries: Option<u64>,
+    pub payment_processed_count_without_smart_retries: Option<u64>,
+    pub payments_success_rate_distribution_without_smart_retries: Option<f64>,
+    pub payments_failure_rate_distribution_without_smart_retries: Option<f64>,
 }
 
 #[derive(Debug, serde::Serialize)]
diff --git a/crates/api_models/src/analytics/payments.rs b/crates/api_models/src/analytics/payments.rs
index c474d47ee63..1120ab092d7 100644
--- a/crates/api_models/src/analytics/payments.rs
+++ b/crates/api_models/src/analytics/payments.rs
@@ -33,6 +33,14 @@ pub struct PaymentFilters {
     pub card_network: Vec<CardNetwork>,
     #[serde(default)]
     pub profile_id: Vec<id_type::ProfileId>,
+    #[serde(default)]
+    pub merchant_id: Vec<id_type::MerchantId>,
+    #[serde(default)]
+    pub card_last_4: Vec<String>,
+    #[serde(default)]
+    pub card_issuer: Vec<String>,
+    #[serde(default)]
+    pub error_reason: Vec<String>,
 }
 
 #[derive(
@@ -68,6 +76,12 @@ pub enum PaymentDimensions {
     ClientVersion,
     ProfileId,
     CardNetwork,
+    MerchantId,
+    #[strum(serialize = "card_last_4")]
+    #[serde(rename = "card_last_4")]
+    CardLast4,
+    CardIssuer,
+    ErrorReason,
 }
 
 #[derive(
@@ -92,6 +106,15 @@ pub enum PaymentMetrics {
     AvgTicketSize,
     RetriesCount,
     ConnectorSuccessRate,
+    SessionizedPaymentSuccessRate,
+    SessionizedPaymentCount,
+    SessionizedPaymentSuccessCount,
+    SessionizedPaymentProcessedAmount,
+    SessionizedAvgTicketSize,
+    SessionizedRetriesCount,
+    SessionizedConnectorSuccessRate,
+    PaymentsDistribution,
+    FailureReasons,
 }
 
 #[derive(Debug, Default, serde::Serialize)]
@@ -159,6 +182,11 @@ pub struct PaymentMetricsBucketIdentifier {
     pub client_source: Option<String>,
     pub client_version: Option<String>,
     pub profile_id: Option<String>,
+    pub card_network: Option<String>,
+    pub merchant_id: Option<String>,
+    pub card_last_4: Option<String>,
+    pub card_issuer: Option<String>,
+    pub error_reason: Option<String>,
     #[serde(rename = "time_range")]
     pub time_bucket: TimeRange,
     // Coz FE sucks
@@ -179,6 +207,11 @@ impl PaymentMetricsBucketIdentifier {
         client_source: Option<String>,
         client_version: Option<String>,
         profile_id: Option<String>,
+        card_network: Option<String>,
+        merchant_id: Option<String>,
+        card_last_4: Option<String>,
+        card_issuer: Option<String>,
+        error_reason: Option<String>,
         normalized_time_range: TimeRange,
     ) -> Self {
         Self {
@@ -191,6 +224,11 @@ impl PaymentMetricsBucketIdentifier {
             client_source,
             client_version,
             profile_id,
+            card_network,
+            merchant_id,
+            card_last_4,
+            card_issuer,
+            error_reason,
             time_bucket: normalized_time_range,
             start_time: normalized_time_range.start_time,
         }
@@ -208,6 +246,11 @@ impl Hash for PaymentMetricsBucketIdentifier {
         self.client_source.hash(state);
         self.client_version.hash(state);
         self.profile_id.hash(state);
+        self.card_network.hash(state);
+        self.merchant_id.hash(state);
+        self.card_last_4.hash(state);
+        self.card_issuer.hash(state);
+        self.error_reason.hash(state);
         self.time_bucket.hash(state);
     }
 }
@@ -228,11 +271,20 @@ pub struct PaymentMetricsBucketValue {
     pub payment_count: Option<u64>,
     pub payment_success_count: Option<u64>,
     pub payment_processed_amount: Option<u64>,
+    pub payment_processed_count: Option<u64>,
+    pub payment_processed_amount_without_smart_retries: Option<u64>,
+    pub payment_processed_count_without_smart_retries: Option<u64>,
     pub avg_ticket_size: Option<f64>,
     pub payment_error_message: Option<Vec<ErrorResult>>,
     pub retries_count: Option<u64>,
     pub retries_amount_processed: Option<u64>,
     pub connector_success_rate: Option<f64>,
+    pub payments_success_rate_distribution: Option<f64>,
+    pub payments_success_rate_distribution_without_smart_retries: Option<f64>,
+    pub payments_failure_rate_distribution: Option<f64>,
+    pub payments_failure_rate_distribution_without_smart_retries: Option<f64>,
+    pub failure_reason_count: Option<u64>,
+    pub failure_reason_count_without_smart_retries: Option<u64>,
 }
 
 #[derive(Debug, serde::Serialize)]
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index 5816df518bd..7272abffbff 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -131,6 +131,7 @@ impl_api_event_type!(
         GetDisputeFilterRequest,
         DisputeFiltersResponse,
         GetDisputeMetricRequest,
+        SankeyResponse,
         OrganizationResponse,
         OrganizationCreateRequest,
         OrganizationUpdateRequest,
@@ -155,6 +156,18 @@ impl<T> ApiEventMetric for MetricsResponse<T> {
     }
 }
 
+impl<T> ApiEventMetric for PaymentsMetricsResponse<T> {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::Miscellaneous)
+    }
+}
+
+impl<T> ApiEventMetric for PaymentIntentsMetricsResponse<T> {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::Miscellaneous)
+    }
+}
+
 #[cfg(all(feature = "v2", feature = "payment_methods_v2"))]
 impl ApiEventMetric for PaymentMethodIntentConfirmInternal {
     fn get_api_event_type(&self) -> Option<ApiEventsType> {
diff --git a/crates/router/src/analytics.rs b/crates/router/src/analytics.rs
index 89159bcff86..aa6db56bb34 100644
--- a/crates/router/src/analytics.rs
+++ b/crates/router/src/analytics.rs
@@ -26,6 +26,7 @@ pub mod routes {
         GetSdkEventMetricRequest, ReportRequest,
     };
     use common_enums::EntityType;
+    use common_utils::types::TimeRange;
     use error_stack::{report, ResultExt};
     use futures::{stream::FuturesUnordered, StreamExt};
 
@@ -142,6 +143,10 @@ pub mod routes {
                             web::resource("filters/disputes")
                                 .route(web::post().to(get_merchant_dispute_filters)),
                         )
+                        .service(
+                            web::resource("metrics/sankey")
+                                .route(web::post().to(get_merchant_sankey)),
+                        )
                         .service(
                             web::scope("/merchant")
                                 .service(
@@ -190,6 +195,10 @@ pub mod routes {
                                 .service(
                                     web::resource("filters/disputes")
                                         .route(web::post().to(get_merchant_dispute_filters)),
+                                )
+                                .service(
+                                    web::resource("metrics/sankey")
+                                        .route(web::post().to(get_merchant_sankey)),
                                 ),
                         )
                         .service(
@@ -232,6 +241,10 @@ pub mod routes {
                                 .service(
                                     web::resource("report/payments")
                                         .route(web::post().to(generate_org_payment_report)),
+                                )
+                                .service(
+                                    web::resource("metrics/sankey")
+                                        .route(web::post().to(get_org_sankey)),
                                 ),
                         )
                         .service(
@@ -290,6 +303,10 @@ pub mod routes {
                                 .service(
                                     web::resource("sdk_event_logs")
                                         .route(web::post().to(get_profile_sdk_events)),
+                                )
+                                .service(
+                                    web::resource("metrics/sankey")
+                                        .route(web::post().to(get_profile_sankey)),
                                 ),
                         ),
                 )
@@ -2277,4 +2294,103 @@ pub mod routes {
         ))
         .await
     }
+
+    pub async fn get_merchant_sankey(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Json<TimeRange>,
+    ) -> impl Responder {
+        let flow = AnalyticsFlow::GetSankey;
+        let payload = json_payload.into_inner();
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            payload,
+            |state, auth: AuthenticationData, req, _| async move {
+                let org_id = auth.merchant_account.get_org_id();
+                let merchant_id = auth.merchant_account.get_id();
+                let auth: AuthInfo = AuthInfo::MerchantLevel {
+                    org_id: org_id.clone(),
+                    merchant_ids: vec![merchant_id.clone()],
+                };
+                analytics::payment_intents::get_sankey(&state.pool, &auth, req)
+                    .await
+                    .map(ApplicationResponse::Json)
+            },
+            &auth::JWTAuth {
+                permission: Permission::Analytics,
+                minimum_entity_level: EntityType::Merchant,
+            },
+            api_locking::LockAction::NotApplicable,
+        ))
+        .await
+    }
+
+    pub async fn get_org_sankey(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Json<TimeRange>,
+    ) -> impl Responder {
+        let flow = AnalyticsFlow::GetSankey;
+        let payload = json_payload.into_inner();
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            payload,
+            |state, auth: AuthenticationData, req, _| async move {
+                let org_id = auth.merchant_account.get_org_id();
+                let auth: AuthInfo = AuthInfo::OrgLevel {
+                    org_id: org_id.clone(),
+                };
+                analytics::payment_intents::get_sankey(&state.pool, &auth, req)
+                    .await
+                    .map(ApplicationResponse::Json)
+            },
+            &auth::JWTAuth {
+                permission: Permission::Analytics,
+                minimum_entity_level: EntityType::Organization,
+            },
+            api_locking::LockAction::NotApplicable,
+        ))
+        .await
+    }
+
+    pub async fn get_profile_sankey(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Json<TimeRange>,
+    ) -> impl Responder {
+        let flow = AnalyticsFlow::GetSankey;
+        let payload = json_payload.into_inner();
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            payload,
+            |state: crate::routes::SessionState, auth: AuthenticationData, req, _| async move {
+                let org_id = auth.merchant_account.get_org_id();
+                let merchant_id = auth.merchant_account.get_id();
+                let profile_id = auth
+                    .profile_id
+                    .ok_or(report!(UserErrors::JwtProfileIdMissing))
+                    .change_context(AnalyticsError::AccessForbiddenError)?;
+                let auth: AuthInfo = AuthInfo::ProfileLevel {
+                    org_id: org_id.clone(),
+                    merchant_id: merchant_id.clone(),
+                    profile_ids: vec![profile_id.clone()],
+                };
+                analytics::payment_intents::get_sankey(&state.pool, &auth, req)
+                    .await
+                    .map(ApplicationResponse::Json)
+            },
+            &auth::JWTAuth {
+                permission: Permission::Analytics,
+                minimum_entity_level: EntityType::Profile,
+            },
+            api_locking::LockAction::NotApplicable,
+        ))
+        .await
+    }
 }

## 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 -->
Adding metrics, filters and APIs for Analytics v2 Dashboard - Payments page
The new Analytics Dashboard will now have the following metrics:
- Sessionizer Payment Intents based Metrics:
  - Total Payments Savings - Amount saved via Smart Retries
  - Payments Success Rate - Overall successful payment intents divided by total intents
  - Payment Lifecycle - Displayed as a Sankey
  - Total Payments Processed - Total payments processed amount and count based on all successful intents
  - Successful Payments Distribution - Overall successful intents divided by total number of intents based on group by
  - Failed Payments Distribution - Overall failed intents divided by total number of intents based on group by

- Sessionizer Payment Attempts based Metrics:
  - Total Payments Processed - Total payments processed amount and count based on all successful attempts
  - Successful Payments Distribution - Overall successful attempts divided by total number of attempts based on group by
  - Failed Payments Distribution - Overall failed attempts divided by total number of attempts based on group by
  - Failure Reasons Distribution - Failure Reason count along with ratio of count / total failed attempts based on group by

- Refunds based Metrics:
  - Total Refunds Processed - Successful refunds

 - Disputes based Metrics:
  - All Disputes - Total number of disputes irrespective of status

The following filters have been added:
- Payment Attempts:
  - Merchant_id
  - Card last 4
  - Error Reason
  - Card Issuer

- Payment Intents:
  - connector
  - authentication_type
  - payment_method
  - payment_method_type
  - card_network
  - merchant id
  - card_last_4
  - error_reason
  - card_issuer

New API endpoints for Payments Lifecycle (Sankey) on different levels:

Payments Lifecycle (Sankey) |  
-- | --
Org Level | /analytics/v1/org/metrics/sankey
Merchant Level | /analytics/v1/merchant/metrics/sankey
Profile Level | /analytics/v1/profile/metrics/sankey

Separated out the existing payment_attempts and payment_intents based metrics and sessionizer_payment_attempts and sessionizer_payment_intents based metrics to support backwards compatibility

### 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).
-->
Enhancing dashboard experience and providing meaningful and important data analytics for the merchants to make informed decisions

## 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)?
-->
Curls for different metrics (which are newly created or modified):
Payment Attempts based metrics:
- sessionized_payment_processed_amount
```bash
curl --location 'http://localhost:8080/analytics/v1/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: test_admin' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyODIxMTk4Nywib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb19BN1F4SjN6TG05Z1NmYkZqUTNpNSJ9.RG85zvOvkm3vikVXppvevenXHkr-cI0nfIhGhiZcMts' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-10-01T23:30:00Z"
        },
        "groupByNames": [
            "connector", "payment_method"
        ],
        "filters": {
            "merchant_id": ["merchant_1726046328"]
        },
        "timeSeries": {
            "granularity": "G_ONEDAY"
        },
        "source": "BATCH",
        "metrics": [
            "sessionized_payment_processed_amount"
        ],
        "delta": true
    }
]'
```
Fields to look out for in the response body:
queryData:
```bash
"payment_processed_amount": 6540,
"payment_processed_count": 1,
"payment_processed_amount_without_smart_retries": 0,
"payment_processed_count_without_smart_retries": null,
```
metaData:
```bash
"total_payment_processed_amount": 117720,
"total_payment_processed_amount_without_smart_retries": 0,
"total_payment_processed_count": 18,
"total_payment_processed_count_without_smart_retries": 0,
```
- payments_distribution
```bash
curl --location 'http://localhost:8080/analytics/v1/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: test_admin' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyODIxMTk4Nywib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb19BN1F4SjN6TG05Z1NmYkZqUTNpNSJ9.RG85zvOvkm3vikVXppvevenXHkr-cI0nfIhGhiZcMts' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-10-01T23:30:00Z"
        },
        "groupByNames": [
            "connector"
        ],
        "filters": {
            "merchant_id": ["merchant_1726046328"]
        },
        "source": "BATCH",
        "metrics": [
            "payments_distribution"
        ],
        "delta": true
    }
]'
```
Fields to look out for in the response body:
queryData:
```bash
"payments_success_rate_distribution": 75.0,
"payments_failure_rate_distribution": 25.0,
```
- failure_reasons
```bash
curl --location 'http://localhost:8080/analytics/v1/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: test_admin' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyODIxMTk4Nywib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb19BN1F4SjN6TG05Z1NmYkZqUTNpNSJ9.RG85zvOvkm3vikVXppvevenXHkr-cI0nfIhGhiZcMts' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-10-01T23:30:00Z"
        },
        "groupByNames": [
            "connector", "error_reason"
        ],
        "filters": {
            "merchant_id": ["merchant_1726046328"]
        },
        "source": "BATCH",
        "metrics": [
            "failure_reasons"
        ],
        "delta": true
    }
]'
```
Fields to look out for in the response body:
queryData:
```bash
"connector": "adyen",
"failure_reason_count": 1,
"failure_reason_count_without_smart_retries": 0,
"error_reason": "Test Failure Reason 4",
```
metaData:
```bash
"total_failure_reasons_count": 7,
"total_failure_reasons_count_without_smart_retries": 0
```

Payment Intents based metrics:
- sessionized_smart_retried_amount
```bash
curl --location 'http://localhost:8080/analytics/v2/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'QueryType: SingleStatTimeseries' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: dev_LZcIA7XeMpnK5satp4CDvjcnHCaeKTBcosyuBBJaknZu1odhFo96cwS0nSdfzuJF' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyODIxMTk4Nywib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb19BN1F4SjN6TG05Z1NmYkZqUTNpNSJ9.RG85zvOvkm3vikVXppvevenXHkr-cI0nfIhGhiZcMts' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-09-26T09:22:30Z"
        },
        "timeSeries": {
            "granularity": "G_ONEDAY"
        },
        "filters": {
            "merchant_id": ["merchant_1726046328"]
        },
        "mode": "ORDER",
        "source": "BATCH",
        "metrics": [
            "sessionized_smart_retried_amount"
        ]
    }
]'
```
Fields to look out for in the response body
queryData:
```bash
"smart_retried_amount": 6540,
"smart_retried_amount_without_smart_retries": 0,
```
metaData
```bash
"total_smart_retried_amount": 6540,
"total_smart_retried_amount_without_smart_retries": 0
```

- sessionized_payments_success_rate

```bash
curl --location 'http://localhost:8080/analytics/v2/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'QueryType: SingleStatTimeseries' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: dev_LZcIA7XeMpnK5satp4CDvjcnHCaeKTBcosyuBBJaknZu1odhFo96cwS0nSdfzuJF' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyODIxMTk4Nywib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb19BN1F4SjN6TG05Z1NmYkZqUTNpNSJ9.RG85zvOvkm3vikVXppvevenXHkr-cI0nfIhGhiZcMts' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-09-26T09:22:30Z"
        },
        "timeSeries": {
            "granularity": "G_ONEDAY"
        },
        "filters": {
            "merchant_id": ["merchant_1726046328"]
        },
        "mode": "ORDER",
        "source": "BATCH",
        "metrics": [
            "sessionized_payments_success_rate"
        ]
    }
]'
```
Fields to look out for in the response body:
queryData:
```bash
"successful_payments": 7,
"successful_payments_without_smart_retries": 7,
"total_payments": 7,
"payments_success_rate": 100.0,
"payments_success_rate_without_smart_retries": 100.0,
```

metaData:
```bash
"total_success_rate": 88.23529411764706,
"total_success_rate_without_smart_retries": 82.3529411764706,
```

- sessionized_payment_processed_amount
```bash
curl --location 'http://localhost:8080/analytics/v2/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'QueryType: SingleStatTimeseries' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: dev_LZcIA7XeMpnK5satp4CDvjcnHCaeKTBcosyuBBJaknZu1odhFo96cwS0nSdfzuJF' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyODczNDQyMywib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb192NXNGb0hlODBPZWlVbElvbm9jTSJ9.716-yn9HhCUU6-1lymVwVWtd82GG4u2m2t_bbcG3cwc' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-10-26T09:22:30Z"
        },
        "timeSeries": {
            "granularity": "G_ONEDAY"
        },
        "filters": {
            "merchant_id": ["merchant_1726046328"]
        },
        "groupByNames": [
            "connector"
        ],
        "mode": "ORDER",
        "source": "BATCH",
        "metrics": [
            "sessionized_payment_processed_amount"
        ]
    }
]'
```
Fields to look out for in the response body:
queryData:
```bash
"payment_processed_amount": 26160,
"payment_processed_count": 4,
"payment_processed_amount_without_smart_retries": 19620,
"payment_processed_count_without_smart_retries": 3,
```
metaData:
```bash
"total_payment_processed_amount": 26160,
"total_payment_processed_amount_without_smart_retries": 19620,
"total_payment_processed_count": 4,
"total_payment_processed_count_without_smart_retries": 3
```

- sessionized_payments_distribution
```bash
curl --location 'http://localhost:8080/analytics/v2/org/metrics/payments' \
--header 'Accept: */*' \
--header 'Accept-Language: en-US,en;q=0.9' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'QueryType: SingleStatTimeseries' \
--header 'Referer: http://localhost:9000/' \
--header 'Sec-Fetch-Dest: empty' \
--header 'Sec-Fetch-Mode: cors' \
--header 'Sec-Fetch-Site: same-site' \
--header 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' \
--header 'api-key: dev_LZcIA7XeMpnK5satp4CDvjcnHCaeKTBcosyuBBJaknZu1odhFo96cwS0nSdfzuJF' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiNTQ5ZTNkMmItMTY5Yi00NzUzLWJmNTQtZDcxMTM2YjRiN2JkIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI2MDQ2MzI4Iiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTcyODczNDQyMywib3JnX2lkIjoib3JnX1ZwU0hPanNZZkR2YWJWWUpnQ0FKIiwicHJvZmlsZV9pZCI6InByb192NXNGb0hlODBPZWlVbElvbm9jTSJ9.716-yn9HhCUU6-1lymVwVWtd82GG4u2m2t_bbcG3cwc' \
--header 'sec-ch-ua: "Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-09-11T00:00:00Z",
            "endTime": "2024-10-26T09:22:30Z"
        },
        "timeSeries": {
            "granularity": "G_ONEDAY"
        },
        "filters": {
            "merchant_id": ["merchant_1726046328"]
        },
        "groupByNames": [
            "connector"
        ],
        "mode": "ORDER",
        "source": "BATCH",
        "metrics": [
            "sessionized_payments_distribution"
        ]
    }
]'
```
Fields to look out for in the response body:
queryData:
```bash
"payments_success_rate_distribution_without_smart_retries": 50.0,
"payments_failure_rate_distribution_without_smart_retries": 16.666666666666668,
```

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

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