Bug: feat(analytics): Add v2 payment analytics



Add analytics based on the payment intent
since merchants rely on payment intents this would help more closely relate to the merchant facing data.

Currently this will support 4 metrics
- successful smart retries
- total smart retries
- smart retried amount
- payment intent count

filters
- status
- currency

group by
- status
- currency
diff --git a/crates/analytics/src/clickhouse.rs b/crates/analytics/src/clickhouse.rs
index ab47397b8af..b455e79b253 100644
--- a/crates/analytics/src/clickhouse.rs
+++ b/crates/analytics/src/clickhouse.rs
@@ -10,6 +10,7 @@ use super::{
     active_payments::metrics::ActivePaymentsMetricRow,
     auth_events::metrics::AuthEventMetricRow,
     health_check::HealthCheck,
+    payment_intents::{filters::PaymentIntentFilterRow, metrics::PaymentIntentMetricRow},
     payments::{
         distribution::PaymentDistributionRow, filters::FilterRow, metrics::PaymentMetricRow,
     },
@@ -157,6 +158,8 @@ where
 impl super::payments::filters::PaymentFilterAnalytics for ClickhouseClient {}
 impl super::payments::metrics::PaymentMetricAnalytics for ClickhouseClient {}
 impl super::payments::distribution::PaymentDistributionAnalytics for ClickhouseClient {}
+impl super::payment_intents::filters::PaymentIntentFilterAnalytics for ClickhouseClient {}
+impl super::payment_intents::metrics::PaymentIntentMetricAnalytics for ClickhouseClient {}
 impl super::refunds::metrics::RefundMetricAnalytics for ClickhouseClient {}
 impl super::refunds::filters::RefundFilterAnalytics for ClickhouseClient {}
 impl super::sdk_events::filters::SdkEventFilterAnalytics for ClickhouseClient {}
@@ -247,6 +250,26 @@ impl TryInto<FilterRow> for serde_json::Value {
     }
 }
 
+impl TryInto<PaymentIntentMetricRow> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<PaymentIntentMetricRow, Self::Error> {
+        serde_json::from_value(self).change_context(ParsingError::StructParseFailure(
+            "Failed to parse PaymentIntentMetricRow in clickhouse results",
+        ))
+    }
+}
+
+impl TryInto<PaymentIntentFilterRow> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<PaymentIntentFilterRow, Self::Error> {
+        serde_json::from_value(self).change_context(ParsingError::StructParseFailure(
+            "Failed to parse PaymentIntentFilterRow in clickhouse results",
+        ))
+    }
+}
+
 impl TryInto<RefundMetricRow> for serde_json::Value {
     type Error = Report<ParsingError>;
 
diff --git a/crates/analytics/src/core.rs b/crates/analytics/src/core.rs
index f3278349748..2c5945f75b5 100644
--- a/crates/analytics/src/core.rs
+++ b/crates/analytics/src/core.rs
@@ -11,6 +11,11 @@ pub async fn get_domain_info(
             download_dimensions: None,
             dimensions: utils::get_payment_dimensions(),
         },
+        AnalyticsDomain::PaymentIntents => GetInfoResponse {
+            metrics: utils::get_payment_intent_metrics_info(),
+            download_dimensions: None,
+            dimensions: utils::get_payment_intent_dimensions(),
+        },
         AnalyticsDomain::Refunds => GetInfoResponse {
             metrics: utils::get_refund_metrics_info(),
             download_dimensions: None,
diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs
index d3db03a6977..10e628475e8 100644
--- a/crates/analytics/src/lib.rs
+++ b/crates/analytics/src/lib.rs
@@ -3,6 +3,7 @@ pub mod core;
 pub mod disputes;
 pub mod errors;
 pub mod metrics;
+pub mod payment_intents;
 pub mod payments;
 mod query;
 pub mod refunds;
@@ -39,6 +40,10 @@ use api_models::analytics::{
     },
     auth_events::{AuthEventMetrics, AuthEventMetricsBucketIdentifier},
     disputes::{DisputeDimensions, DisputeFilters, DisputeMetrics, DisputeMetricsBucketIdentifier},
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetrics,
+        PaymentIntentMetricsBucketIdentifier,
+    },
     payments::{PaymentDimensions, PaymentFilters, PaymentMetrics, PaymentMetricsBucketIdentifier},
     refunds::{RefundDimensions, RefundFilters, RefundMetrics, RefundMetricsBucketIdentifier},
     sdk_events::{
@@ -60,6 +65,7 @@ use strum::Display;
 use self::{
     active_payments::metrics::{ActivePaymentsMetric, ActivePaymentsMetricRow},
     auth_events::metrics::{AuthEventMetric, AuthEventMetricRow},
+    payment_intents::metrics::{PaymentIntentMetric, PaymentIntentMetricRow},
     payments::{
         distribution::{PaymentDistribution, PaymentDistributionRow},
         metrics::{PaymentMetric, PaymentMetricRow},
@@ -313,6 +319,111 @@ impl AnalyticsProvider {
         .await
     }
 
+    pub async fn get_payment_intent_metrics(
+        &self,
+        metric: &PaymentIntentMetrics,
+        dimensions: &[PaymentIntentDimensions],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+    ) -> types::MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>
+    {
+        // Metrics to get the fetch time for each payment intent metric
+        metrics::request::record_operation_time(
+            async {
+                match self {
+                        Self::Sqlx(pool) => {
+                        metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                pool,
+                            )
+                            .await
+                    }
+                                        Self::Clickhouse(pool) => {
+                        metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                pool,
+                            )
+                            .await
+                    }
+                                    Self::CombinedCkh(sqlx_pool, ckh_pool) => {
+                        let (ckh_result, sqlx_result) = tokio::join!(metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                ckh_pool,
+                            ),
+                            metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                sqlx_pool,
+                            ));
+                        match (&sqlx_result, &ckh_result) {
+                            (Ok(ref sqlx_res), Ok(ref ckh_res)) if sqlx_res != ckh_res => {
+                                router_env::logger::error!(clickhouse_result=?ckh_res, postgres_result=?sqlx_res, "Mismatch between clickhouse & postgres payment intents analytics metrics")
+                            },
+                            _ => {}
+
+                        };
+
+                        ckh_result
+                    }
+                                    Self::CombinedSqlx(sqlx_pool, ckh_pool) => {
+                        let (ckh_result, sqlx_result) = tokio::join!(metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                ckh_pool,
+                            ),
+                            metric
+                            .load_metrics(
+                                dimensions,
+                                merchant_id,
+                                filters,
+                                granularity,
+                                time_range,
+                                sqlx_pool,
+                            ));
+                        match (&sqlx_result, &ckh_result) {
+                            (Ok(ref sqlx_res), Ok(ref ckh_res)) if sqlx_res != ckh_res => {
+                                router_env::logger::error!(clickhouse_result=?ckh_res, postgres_result=?sqlx_res, "Mismatch between clickhouse & postgres payment intents analytics metrics")
+                            },
+                            _ => {}
+
+                        };
+
+                        sqlx_result
+                    }
+                }
+            },
+            &metrics::METRIC_FETCH_TIME,
+            metric,
+            self,
+        )
+        .await
+    }
+
     pub async fn get_refund_metrics(
         &self,
         metric: &RefundMetrics,
@@ -756,11 +867,13 @@ pub struct ReportConfig {
 pub enum AnalyticsFlow {
     GetInfo,
     GetPaymentMetrics,
+    GetPaymentIntentMetrics,
     GetRefundsMetrics,
     GetSdkMetrics,
     GetAuthMetrics,
     GetActivePaymentsMetrics,
     GetPaymentFilters,
+    GetPaymentIntentFilters,
     GetRefundFilters,
     GetSdkEventFilters,
     GetApiEvents,
diff --git a/crates/analytics/src/payment_intents.rs b/crates/analytics/src/payment_intents.rs
new file mode 100644
index 00000000000..449dd94788c
--- /dev/null
+++ b/crates/analytics/src/payment_intents.rs
@@ -0,0 +1,13 @@
+pub mod accumulator;
+mod core;
+pub mod filters;
+pub mod metrics;
+pub mod types;
+pub use accumulator::{PaymentIntentMetricAccumulator, PaymentIntentMetricsAccumulator};
+
+pub trait PaymentIntentAnalytics:
+    metrics::PaymentIntentMetricAnalytics + filters::PaymentIntentFilterAnalytics
+{
+}
+
+pub use self::core::{get_filters, get_metrics};
diff --git a/crates/analytics/src/payment_intents/accumulator.rs b/crates/analytics/src/payment_intents/accumulator.rs
new file mode 100644
index 00000000000..8fd98a1e73c
--- /dev/null
+++ b/crates/analytics/src/payment_intents/accumulator.rs
@@ -0,0 +1,90 @@
+use api_models::analytics::payment_intents::PaymentIntentMetricsBucketValue;
+use bigdecimal::ToPrimitive;
+
+use super::metrics::PaymentIntentMetricRow;
+
+#[derive(Debug, Default)]
+pub struct PaymentIntentMetricsAccumulator {
+    pub successful_smart_retries: CountAccumulator,
+    pub total_smart_retries: CountAccumulator,
+    pub smart_retried_amount: SumAccumulator,
+    pub payment_intent_count: CountAccumulator,
+}
+
+#[derive(Debug, Default)]
+pub struct ErrorDistributionRow {
+    pub count: i64,
+    pub total: i64,
+    pub error_message: String,
+}
+
+#[derive(Debug, Default)]
+pub struct ErrorDistributionAccumulator {
+    pub error_vec: Vec<ErrorDistributionRow>,
+}
+
+#[derive(Debug, Default)]
+#[repr(transparent)]
+pub struct CountAccumulator {
+    pub count: Option<i64>,
+}
+
+pub trait PaymentIntentMetricAccumulator {
+    type MetricOutput;
+
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow);
+
+    fn collect(self) -> Self::MetricOutput;
+}
+
+#[derive(Debug, Default)]
+#[repr(transparent)]
+pub struct SumAccumulator {
+    pub total: Option<i64>,
+}
+
+impl PaymentIntentMetricAccumulator for CountAccumulator {
+    type MetricOutput = Option<u64>;
+    #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
+        self.count = match (self.count, 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 {
+        self.count.and_then(|i| u64::try_from(i).ok())
+    }
+}
+
+impl PaymentIntentMetricAccumulator for SumAccumulator {
+    type MetricOutput = Option<u64>;
+    #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &PaymentIntentMetricRow) {
+        self.total = match (
+            self.total,
+            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),
+        }
+    }
+    #[inline]
+    fn collect(self) -> Self::MetricOutput {
+        self.total.and_then(|i| u64::try_from(i).ok())
+    }
+}
+
+impl PaymentIntentMetricsAccumulator {
+    pub fn collect(self) -> PaymentIntentMetricsBucketValue {
+        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(),
+            payment_intent_count: self.payment_intent_count.collect(),
+        }
+    }
+}
diff --git a/crates/analytics/src/payment_intents/core.rs b/crates/analytics/src/payment_intents/core.rs
new file mode 100644
index 00000000000..e3932baaf07
--- /dev/null
+++ b/crates/analytics/src/payment_intents/core.rs
@@ -0,0 +1,226 @@
+#![allow(dead_code)]
+use std::collections::HashMap;
+
+use api_models::analytics::{
+    payment_intents::{
+        MetricsBucketResponse, PaymentIntentDimensions, PaymentIntentMetrics,
+        PaymentIntentMetricsBucketIdentifier,
+    },
+    AnalyticsMetadata, GetPaymentIntentFiltersRequest, GetPaymentIntentMetricRequest,
+    MetricsResponse, PaymentIntentFilterValue, PaymentIntentFiltersResponse,
+};
+use common_utils::errors::CustomResult;
+use error_stack::ResultExt;
+use router_env::{
+    instrument, logger,
+    metrics::add_attributes,
+    tracing::{self, Instrument},
+};
+
+use super::{
+    filters::{get_payment_intent_filter_for_dimension, PaymentIntentFilterRow},
+    metrics::PaymentIntentMetricRow,
+    PaymentIntentMetricsAccumulator,
+};
+use crate::{
+    errors::{AnalyticsError, AnalyticsResult},
+    metrics,
+    payment_intents::PaymentIntentMetricAccumulator,
+    AnalyticsProvider,
+};
+
+#[derive(Debug)]
+pub enum TaskType {
+    MetricTask(
+        PaymentIntentMetrics,
+        CustomResult<
+            Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+            AnalyticsError,
+        >,
+    ),
+}
+
+#[instrument(skip_all)]
+pub async fn get_metrics(
+    pool: &AnalyticsProvider,
+    merchant_id: &str,
+    req: GetPaymentIntentMetricRequest,
+) -> AnalyticsResult<MetricsResponse<MetricsBucketResponse>> {
+    let mut metrics_accumulator: HashMap<
+        PaymentIntentMetricsBucketIdentifier,
+        PaymentIntentMetricsAccumulator,
+    > = HashMap::new();
+
+    let mut set = tokio::task::JoinSet::new();
+    for metric_type in req.metrics.iter().cloned() {
+        let req = req.clone();
+        let pool = pool.clone();
+        let task_span = tracing::debug_span!(
+            "analytics_payment_intents_metrics_query",
+            payment_metric = metric_type.as_ref()
+        );
+
+        // TODO: lifetime issues with joinset,
+        // can be optimized away if joinset lifetime requirements are relaxed
+        let merchant_id_scoped = merchant_id.to_owned();
+        set.spawn(
+            async move {
+                let data = pool
+                    .get_payment_intent_metrics(
+                        &metric_type,
+                        &req.group_by_names.clone(),
+                        &merchant_id_scoped,
+                        &req.filters,
+                        &req.time_series.map(|t| t.granularity),
+                        &req.time_range,
+                    )
+                    .await
+                    .change_context(AnalyticsError::UnknownError);
+                TaskType::MetricTask(metric_type, data)
+            }
+            .instrument(task_span),
+        );
+    }
+
+    while let Some(task_type) = set
+        .join_next()
+        .await
+        .transpose()
+        .change_context(AnalyticsError::UnknownError)?
+    {
+        match task_type {
+            TaskType::MetricTask(metric, data) => {
+                let data = data?;
+                let attributes = &add_attributes([
+                    ("metric_type", metric.to_string()),
+                    ("source", pool.to_string()),
+                ]);
+
+                let value = u64::try_from(data.len());
+                if let Ok(val) = value {
+                    metrics::BUCKETS_FETCHED.record(&metrics::CONTEXT, val, attributes);
+                    logger::debug!("Attributes: {:?}, Buckets fetched: {}", attributes, val);
+                }
+
+                for (id, value) in data {
+                    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
+                            .total_smart_retries
+                            .add_metrics_bucket(&value),
+                        PaymentIntentMetrics::SmartRetriedAmount => metrics_builder
+                            .smart_retried_amount
+                            .add_metrics_bucket(&value),
+                        PaymentIntentMetrics::PaymentIntentCount => metrics_builder
+                            .payment_intent_count
+                            .add_metrics_bucket(&value),
+                    }
+                }
+
+                logger::debug!(
+                    "Analytics Accumulated Results: metric: {}, results: {:#?}",
+                    metric,
+                    metrics_accumulator
+                );
+            }
+        }
+    }
+
+    let query_data: Vec<MetricsBucketResponse> = metrics_accumulator
+        .into_iter()
+        .map(|(id, val)| MetricsBucketResponse {
+            values: val.collect(),
+            dimensions: id,
+        })
+        .collect();
+
+    Ok(MetricsResponse {
+        query_data,
+        meta_data: [AnalyticsMetadata {
+            current_time_range: req.time_range,
+        }],
+    })
+}
+
+pub async fn get_filters(
+    pool: &AnalyticsProvider,
+    req: GetPaymentIntentFiltersRequest,
+    merchant_id: &String,
+) -> AnalyticsResult<PaymentIntentFiltersResponse> {
+    let mut res = PaymentIntentFiltersResponse::default();
+
+    for dim in req.group_by_names {
+        let values = match pool {
+                        AnalyticsProvider::Sqlx(pool) => {
+                get_payment_intent_filter_for_dimension(dim, merchant_id, &req.time_range, pool)
+                    .await
+            }
+                        AnalyticsProvider::Clickhouse(pool) => {
+                get_payment_intent_filter_for_dimension(dim, merchant_id, &req.time_range, pool)
+                    .await
+            }
+                    AnalyticsProvider::CombinedCkh(sqlx_poll, ckh_pool) => {
+                let ckh_result = get_payment_intent_filter_for_dimension(
+                    dim,
+                    merchant_id,
+                    &req.time_range,
+                    ckh_pool,
+                )
+                .await;
+                let sqlx_result = get_payment_intent_filter_for_dimension(
+                    dim,
+                    merchant_id,
+                    &req.time_range,
+                    sqlx_poll,
+                )
+                .await;
+                match (&sqlx_result, &ckh_result) {
+                    (Ok(ref sqlx_res), Ok(ref ckh_res)) if sqlx_res != ckh_res => {
+                        router_env::logger::error!(clickhouse_result=?ckh_res, postgres_result=?sqlx_res, "Mismatch between clickhouse & postgres payment intents analytics filters")
+                    },
+                    _ => {}
+                };
+                ckh_result
+            }
+                    AnalyticsProvider::CombinedSqlx(sqlx_poll, ckh_pool) => {
+                let ckh_result = get_payment_intent_filter_for_dimension(
+                    dim,
+                    merchant_id,
+                    &req.time_range,
+                    ckh_pool,
+                )
+                .await;
+                let sqlx_result = get_payment_intent_filter_for_dimension(
+                    dim,
+                    merchant_id,
+                    &req.time_range,
+                    sqlx_poll,
+                )
+                .await;
+                match (&sqlx_result, &ckh_result) {
+                    (Ok(ref sqlx_res), Ok(ref ckh_res)) if sqlx_res != ckh_res => {
+                        router_env::logger::error!(clickhouse_result=?ckh_res, postgres_result=?sqlx_res, "Mismatch between clickhouse & postgres payment intents analytics filters")
+                    },
+                    _ => {}
+                };
+                sqlx_result
+            }
+        }
+        .change_context(AnalyticsError::UnknownError)?
+        .into_iter()
+        .filter_map(|fil: PaymentIntentFilterRow| match dim {
+            PaymentIntentDimensions::PaymentIntentStatus => fil.status.map(|i| i.as_ref().to_string()),
+            PaymentIntentDimensions::Currency => fil.currency.map(|i| i.as_ref().to_string()),
+        })
+        .collect::<Vec<String>>();
+        res.query_data.push(PaymentIntentFilterValue {
+            dimension: dim,
+            values,
+        })
+    }
+    Ok(res)
+}
diff --git a/crates/analytics/src/payment_intents/filters.rs b/crates/analytics/src/payment_intents/filters.rs
new file mode 100644
index 00000000000..1a74cfd510e
--- /dev/null
+++ b/crates/analytics/src/payment_intents/filters.rs
@@ -0,0 +1,56 @@
+use api_models::analytics::{payment_intents::PaymentIntentDimensions, Granularity, TimeRange};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums::{Currency, IntentStatus};
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use crate::{
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, ToSql, Window},
+    types::{
+        AnalyticsCollection, AnalyticsDataSource, DBEnumWrapper, FiltersError, FiltersResult,
+        LoadRow,
+    },
+};
+
+pub trait PaymentIntentFilterAnalytics: LoadRow<PaymentIntentFilterRow> {}
+
+pub async fn get_payment_intent_filter_for_dimension<T>(
+    dimension: PaymentIntentDimensions,
+    merchant: &String,
+    time_range: &TimeRange,
+    pool: &T,
+) -> FiltersResult<Vec<PaymentIntentFilterRow>>
+where
+    T: AnalyticsDataSource + PaymentIntentFilterAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::PaymentIntent);
+
+    query_builder.add_select_column(dimension).switch()?;
+    time_range
+        .set_filter_clause(&mut query_builder)
+        .attach_printable("Error filtering time range")
+        .switch()?;
+
+    query_builder
+        .add_filter_clause("merchant_id", merchant)
+        .switch()?;
+
+    query_builder.set_distinct();
+
+    query_builder
+        .execute_query::<PaymentIntentFilterRow, _>(pool)
+        .await
+        .change_context(FiltersError::QueryBuildingError)?
+        .change_context(FiltersError::QueryExecutionFailure)
+}
+
+#[derive(Debug, serde::Serialize, Eq, PartialEq, serde::Deserialize)]
+pub struct PaymentIntentFilterRow {
+    pub status: Option<DBEnumWrapper<IntentStatus>>,
+    pub currency: Option<DBEnumWrapper<Currency>>,
+}
diff --git a/crates/analytics/src/payment_intents/metrics.rs b/crates/analytics/src/payment_intents/metrics.rs
new file mode 100644
index 00000000000..3a0cbbc85db
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics.rs
@@ -0,0 +1,126 @@
+use api_models::analytics::{
+    payment_intents::{
+        PaymentIntentDimensions, PaymentIntentFilters, PaymentIntentMetrics,
+        PaymentIntentMetricsBucketIdentifier,
+    },
+    Granularity, TimeRange,
+};
+use diesel_models::enums as storage_enums;
+use time::PrimitiveDateTime;
+
+use crate::{
+    query::{Aggregate, GroupByClause, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, DBEnumWrapper, LoadRow, MetricsResult},
+};
+
+mod payment_intent_count;
+mod smart_retried_amount;
+mod successful_smart_retries;
+mod total_smart_retries;
+
+use payment_intent_count::PaymentIntentCount;
+use smart_retried_amount::SmartRetriedAmount;
+use successful_smart_retries::SuccessfulSmartRetries;
+use total_smart_retries::TotalSmartRetries;
+
+#[derive(Debug, PartialEq, Eq, serde::Deserialize)]
+pub struct PaymentIntentMetricRow {
+    pub status: Option<DBEnumWrapper<storage_enums::IntentStatus>>,
+    pub currency: Option<DBEnumWrapper<storage_enums::Currency>>,
+    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>,
+}
+
+pub trait PaymentIntentMetricAnalytics: LoadRow<PaymentIntentMetricRow> {}
+
+#[async_trait::async_trait]
+pub trait PaymentIntentMetric<T>
+where
+    T: AnalyticsDataSource + PaymentIntentMetricAnalytics,
+{
+    async fn load_metrics(
+        &self,
+        dimensions: &[PaymentIntentDimensions],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>>;
+}
+
+#[async_trait::async_trait]
+impl<T> PaymentIntentMetric<T> for PaymentIntentMetrics
+where
+    T: AnalyticsDataSource + 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],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>> {
+        match self {
+            Self::SuccessfulSmartRetries => {
+                SuccessfulSmartRetries
+                    .load_metrics(
+                        dimensions,
+                        merchant_id,
+                        filters,
+                        granularity,
+                        time_range,
+                        pool,
+                    )
+                    .await
+            }
+            Self::TotalSmartRetries => {
+                TotalSmartRetries
+                    .load_metrics(
+                        dimensions,
+                        merchant_id,
+                        filters,
+                        granularity,
+                        time_range,
+                        pool,
+                    )
+                    .await
+            }
+            Self::SmartRetriedAmount => {
+                SmartRetriedAmount
+                    .load_metrics(
+                        dimensions,
+                        merchant_id,
+                        filters,
+                        granularity,
+                        time_range,
+                        pool,
+                    )
+                    .await
+            }
+            Self::PaymentIntentCount => {
+                PaymentIntentCount
+                    .load_metrics(
+                        dimensions,
+                        merchant_id,
+                        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
new file mode 100644
index 00000000000..0f235375c4f
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/payment_intent_count.rs
@@ -0,0 +1,121 @@
+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::{
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(super) 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],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntent);
+
+        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()?;
+
+        query_builder
+            .add_filter_clause("merchant_id", merchant_id)
+            .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),
+                        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<
+                Vec<(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
new file mode 100644
index 00000000000..470a0e66867
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/smart_retried_amount.rs
@@ -0,0 +1,131 @@
+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::{
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(super) 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],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntent);
+
+        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::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()?;
+
+        query_builder
+            .add_filter_clause("merchant_id", merchant_id)
+            .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),
+                        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<
+                Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/successful_smart_retries.rs b/crates/analytics/src/payment_intents/metrics/successful_smart_retries.rs
new file mode 100644
index 00000000000..292062d1e10
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/successful_smart_retries.rs
@@ -0,0 +1,130 @@
+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::{
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(super) 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],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntent);
+
+        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()?;
+
+        query_builder
+            .add_filter_clause("merchant_id", merchant_id)
+            .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),
+                        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<
+                Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/metrics/total_smart_retries.rs b/crates/analytics/src/payment_intents/metrics/total_smart_retries.rs
new file mode 100644
index 00000000000..d5a3d142baf
--- /dev/null
+++ b/crates/analytics/src/payment_intents/metrics/total_smart_retries.rs
@@ -0,0 +1,125 @@
+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::{
+    query::{
+        Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql,
+        Window,
+    },
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(super) 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],
+        merchant_id: &str,
+        filters: &PaymentIntentFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::PaymentIntent);
+
+        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()?;
+
+        query_builder
+            .add_filter_clause("merchant_id", merchant_id)
+            .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),
+                        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<
+                Vec<(PaymentIntentMetricsBucketIdentifier, PaymentIntentMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/payment_intents/types.rs b/crates/analytics/src/payment_intents/types.rs
new file mode 100644
index 00000000000..9b1e7b8674d
--- /dev/null
+++ b/crates/analytics/src/payment_intents/types.rs
@@ -0,0 +1,30 @@
+use api_models::analytics::payment_intents::{PaymentIntentDimensions, PaymentIntentFilters};
+use error_stack::ResultExt;
+
+use crate::{
+    query::{QueryBuilder, QueryFilter, QueryResult, ToSql},
+    types::{AnalyticsCollection, AnalyticsDataSource},
+};
+
+impl<T> QueryFilter<T> for PaymentIntentFilters
+where
+    T: AnalyticsDataSource,
+    AnalyticsCollection: ToSql<T>,
+{
+    fn set_filter_clause(&self, builder: &mut QueryBuilder<T>) -> QueryResult<()> {
+        if !self.status.is_empty() {
+            builder
+                .add_filter_in_range_clause(
+                    PaymentIntentDimensions::PaymentIntentStatus,
+                    &self.status,
+                )
+                .attach_printable("Error adding payment intent status filter")?;
+        }
+        if !self.currency.is_empty() {
+            builder
+                .add_filter_in_range_clause(PaymentIntentDimensions::Currency, &self.currency)
+                .attach_printable("Error adding currency filter")?;
+        }
+        Ok(())
+    }
+}
diff --git a/crates/analytics/src/payments/metrics/retries_count.rs b/crates/analytics/src/payments/metrics/retries_count.rs
index 87d80c87fb4..3c4580d37a7 100644
--- a/crates/analytics/src/payments/metrics/retries_count.rs
+++ b/crates/analytics/src/payments/metrics/retries_count.rs
@@ -1,6 +1,9 @@
-use api_models::analytics::{
-    payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
-    Granularity, TimeRange,
+use api_models::{
+    analytics::{
+        payments::{PaymentDimensions, PaymentFilters, PaymentMetricsBucketIdentifier},
+        Granularity, TimeRange,
+    },
+    enums::IntentStatus,
 };
 use common_utils::errors::ReportSwitchExt;
 use error_stack::ResultExt;
@@ -70,7 +73,7 @@ where
             .add_custom_filter_clause("attempt_count", "1", FilterTypes::Gt)
             .switch()?;
         query_builder
-            .add_custom_filter_clause("status", "succeeded", FilterTypes::Equal)
+            .add_custom_filter_clause("status", IntentStatus::Succeeded, FilterTypes::Equal)
             .switch()?;
         time_range
             .set_filter_clause(&mut query_builder)
diff --git a/crates/analytics/src/query.rs b/crates/analytics/src/query.rs
index 2fda8fc57cd..a257fedc09d 100644
--- a/crates/analytics/src/query.rs
+++ b/crates/analytics/src/query.rs
@@ -6,14 +6,15 @@ use api_models::{
         api_event::ApiEventDimensions,
         auth_events::AuthEventFlows,
         disputes::DisputeDimensions,
+        payment_intents::PaymentIntentDimensions,
         payments::{PaymentDimensions, PaymentDistributions},
         refunds::{RefundDimensions, RefundType},
         sdk_events::{SdkEventDimensions, SdkEventNames},
         Granularity,
     },
     enums::{
-        AttemptStatus, AuthenticationType, Connector, Currency, DisputeStage, PaymentMethod,
-        PaymentMethodType,
+        AttemptStatus, AuthenticationType, Connector, Currency, DisputeStage, IntentStatus,
+        PaymentMethod, PaymentMethodType,
     },
     refunds::RefundStatus,
 };
@@ -369,8 +370,10 @@ impl_to_sql_for_to_string!(
     String,
     &str,
     &PaymentDimensions,
+    &PaymentIntentDimensions,
     &RefundDimensions,
     PaymentDimensions,
+    PaymentIntentDimensions,
     &PaymentDistributions,
     RefundDimensions,
     PaymentMethod,
@@ -378,6 +381,7 @@ impl_to_sql_for_to_string!(
     AuthenticationType,
     Connector,
     AttemptStatus,
+    IntentStatus,
     RefundStatus,
     storage_enums::RefundStatus,
     Currency,
diff --git a/crates/analytics/src/sqlx.rs b/crates/analytics/src/sqlx.rs
index 76ad9c254be..6a4faf50eb8 100644
--- a/crates/analytics/src/sqlx.rs
+++ b/crates/analytics/src/sqlx.rs
@@ -9,7 +9,7 @@ use common_utils::{
     DbConnectionParams,
 };
 use diesel_models::enums::{
-    AttemptStatus, AuthenticationType, Currency, PaymentMethod, RefundStatus,
+    AttemptStatus, AuthenticationType, Currency, IntentStatus, PaymentMethod, RefundStatus,
 };
 use error_stack::ResultExt;
 use sqlx::{
@@ -87,6 +87,7 @@ macro_rules! db_type {
 db_type!(Currency);
 db_type!(AuthenticationType);
 db_type!(AttemptStatus);
+db_type!(IntentStatus);
 db_type!(PaymentMethod, TEXT);
 db_type!(RefundStatus);
 db_type!(RefundType);
@@ -143,6 +144,8 @@ where
 impl super::payments::filters::PaymentFilterAnalytics for SqlxClient {}
 impl super::payments::metrics::PaymentMetricAnalytics for SqlxClient {}
 impl super::payments::distribution::PaymentDistributionAnalytics for SqlxClient {}
+impl super::payment_intents::filters::PaymentIntentFilterAnalytics for SqlxClient {}
+impl super::payment_intents::metrics::PaymentIntentMetricAnalytics for SqlxClient {}
 impl super::refunds::metrics::RefundMetricAnalytics for SqlxClient {}
 impl super::refunds::filters::RefundFilterAnalytics for SqlxClient {}
 impl super::disputes::filters::DisputeFilterAnalytics for SqlxClient {}
@@ -429,6 +432,60 @@ impl<'a> FromRow<'a, PgRow> for super::payments::filters::FilterRow {
     }
 }
 
+impl<'a> FromRow<'a, PgRow> for super::payment_intents::metrics::PaymentIntentMetricRow {
+    fn from_row(row: &'a PgRow) -> sqlx::Result<Self> {
+        let status: Option<DBEnumWrapper<IntentStatus>> =
+            row.try_get("status").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let currency: Option<DBEnumWrapper<Currency>> =
+            row.try_get("currency").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),
+        })?;
+        let count: Option<i64> = row.try_get("count").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")?
+            .and_then(|dt| dt.replace_millisecond(0).ok());
+        let end_bucket: Option<PrimitiveDateTime> = row
+            .try_get::<Option<PrimitiveDateTime>, _>("end_bucket")?
+            .and_then(|dt| dt.replace_millisecond(0).ok());
+        Ok(Self {
+            status,
+            currency,
+            total,
+            count,
+            start_bucket,
+            end_bucket,
+        })
+    }
+}
+
+impl<'a> FromRow<'a, PgRow> for super::payment_intents::filters::PaymentIntentFilterRow {
+    fn from_row(row: &'a PgRow) -> sqlx::Result<Self> {
+        let status: Option<DBEnumWrapper<IntentStatus>> =
+            row.try_get("status").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        let currency: Option<DBEnumWrapper<Currency>> =
+            row.try_get("currency").or_else(|e| match e {
+                ColumnNotFound(_) => Ok(Default::default()),
+                e => Err(e),
+            })?;
+        Ok(Self { status, currency })
+    }
+}
+
 impl<'a> FromRow<'a, PgRow> for super::refunds::filters::RefundFilterRow {
     fn from_row(row: &'a PgRow) -> sqlx::Result<Self> {
         let currency: Option<DBEnumWrapper<Currency>> =
diff --git a/crates/analytics/src/types.rs b/crates/analytics/src/types.rs
index 5370fbc25ac..816c77fd304 100644
--- a/crates/analytics/src/types.rs
+++ b/crates/analytics/src/types.rs
@@ -15,6 +15,7 @@ use crate::errors::AnalyticsError;
 pub enum AnalyticsDomain {
     Payments,
     Refunds,
+    PaymentIntents,
     AuthEvents,
     SdkEvents,
     ApiEvents,
diff --git a/crates/analytics/src/utils.rs b/crates/analytics/src/utils.rs
index 0afe9bd6c5e..3955a8c1dfe 100644
--- a/crates/analytics/src/utils.rs
+++ b/crates/analytics/src/utils.rs
@@ -2,6 +2,7 @@ use api_models::analytics::{
     api_event::{ApiEventDimensions, ApiEventMetrics},
     auth_events::AuthEventMetrics,
     disputes::{DisputeDimensions, DisputeMetrics},
+    payment_intents::{PaymentIntentDimensions, PaymentIntentMetrics},
     payments::{PaymentDimensions, PaymentMetrics},
     refunds::{RefundDimensions, RefundMetrics},
     sdk_events::{SdkEventDimensions, SdkEventMetrics},
@@ -13,6 +14,10 @@ pub fn get_payment_dimensions() -> Vec<NameDescription> {
     PaymentDimensions::iter().map(Into::into).collect()
 }
 
+pub fn get_payment_intent_dimensions() -> Vec<NameDescription> {
+    PaymentIntentDimensions::iter().map(Into::into).collect()
+}
+
 pub fn get_refund_dimensions() -> Vec<NameDescription> {
     RefundDimensions::iter().map(Into::into).collect()
 }
@@ -29,6 +34,10 @@ pub fn get_payment_metrics_info() -> Vec<NameDescription> {
     PaymentMetrics::iter().map(Into::into).collect()
 }
 
+pub fn get_payment_intent_metrics_info() -> Vec<NameDescription> {
+    PaymentIntentMetrics::iter().map(Into::into).collect()
+}
+
 pub fn get_refund_metrics_info() -> Vec<NameDescription> {
     RefundMetrics::iter().map(Into::into).collect()
 }
diff --git a/crates/api_models/src/analytics.rs b/crates/api_models/src/analytics.rs
index 491420cfc02..85a9c3ded09 100644
--- a/crates/api_models/src/analytics.rs
+++ b/crates/api_models/src/analytics.rs
@@ -8,6 +8,7 @@ use self::{
     api_event::{ApiEventDimensions, ApiEventMetrics},
     auth_events::AuthEventMetrics,
     disputes::{DisputeDimensions, DisputeMetrics},
+    payment_intents::{PaymentIntentDimensions, PaymentIntentMetrics},
     payments::{PaymentDimensions, PaymentDistributions, PaymentMetrics},
     refunds::{RefundDimensions, RefundMetrics},
     sdk_events::{SdkEventDimensions, SdkEventMetrics},
@@ -20,6 +21,7 @@ pub mod auth_events;
 pub mod connector_events;
 pub mod disputes;
 pub mod outgoing_webhook_event;
+pub mod payment_intents;
 pub mod payments;
 pub mod refunds;
 pub mod sdk_events;
@@ -114,6 +116,20 @@ pub struct GenerateReportRequest {
     pub email: Secret<String, EmailStrategy>,
 }
 
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetPaymentIntentMetricRequest {
+    pub time_series: Option<TimeSeries>,
+    pub time_range: TimeRange,
+    #[serde(default)]
+    pub group_by_names: Vec<PaymentIntentDimensions>,
+    #[serde(default)]
+    pub filters: payment_intents::PaymentIntentFilters,
+    pub metrics: HashSet<PaymentIntentMetrics>,
+    #[serde(default)]
+    pub delta: bool,
+}
+
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GetRefundMetricRequest {
@@ -187,6 +203,27 @@ pub struct FilterValue {
     pub values: Vec<String>,
 }
 
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetPaymentIntentFiltersRequest {
+    pub time_range: TimeRange,
+    #[serde(default)]
+    pub group_by_names: Vec<PaymentIntentDimensions>,
+}
+
+#[derive(Debug, Default, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PaymentIntentFiltersResponse {
+    pub query_data: Vec<PaymentIntentFilterValue>,
+}
+
+#[derive(Debug, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PaymentIntentFilterValue {
+    pub dimension: PaymentIntentDimensions,
+    pub values: Vec<String>,
+}
+
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 
diff --git a/crates/api_models/src/analytics/payment_intents.rs b/crates/api_models/src/analytics/payment_intents.rs
new file mode 100644
index 00000000000..232c1719047
--- /dev/null
+++ b/crates/api_models/src/analytics/payment_intents.rs
@@ -0,0 +1,151 @@
+use std::{
+    collections::hash_map::DefaultHasher,
+    hash::{Hash, Hasher},
+};
+
+use super::{NameDescription, TimeRange};
+use crate::enums::{Currency, IntentStatus};
+
+#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
+pub struct PaymentIntentFilters {
+    #[serde(default)]
+    pub status: Vec<IntentStatus>,
+    pub currency: Vec<Currency>,
+}
+
+#[derive(
+    Debug,
+    serde::Serialize,
+    serde::Deserialize,
+    strum::AsRefStr,
+    PartialEq,
+    PartialOrd,
+    Eq,
+    Ord,
+    strum::Display,
+    strum::EnumIter,
+    Clone,
+    Copy,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum PaymentIntentDimensions {
+    #[strum(serialize = "status")]
+    #[serde(rename = "status")]
+    PaymentIntentStatus,
+    Currency,
+}
+
+#[derive(
+    Clone,
+    Debug,
+    Hash,
+    PartialEq,
+    Eq,
+    serde::Serialize,
+    serde::Deserialize,
+    strum::Display,
+    strum::EnumIter,
+    strum::AsRefStr,
+)]
+#[strum(serialize_all = "snake_case")]
+#[serde(rename_all = "snake_case")]
+pub enum PaymentIntentMetrics {
+    SuccessfulSmartRetries,
+    TotalSmartRetries,
+    SmartRetriedAmount,
+    PaymentIntentCount,
+}
+
+#[derive(Debug, Default, serde::Serialize)]
+pub struct ErrorResult {
+    pub reason: String,
+    pub count: i64,
+    pub percentage: f64,
+}
+
+pub mod metric_behaviour {
+    pub struct SuccessfulSmartRetries;
+    pub struct TotalSmartRetries;
+    pub struct SmartRetriedAmount;
+    pub struct PaymentIntentCount;
+}
+
+impl From<PaymentIntentMetrics> for NameDescription {
+    fn from(value: PaymentIntentMetrics) -> Self {
+        Self {
+            name: value.to_string(),
+            desc: String::new(),
+        }
+    }
+}
+
+impl From<PaymentIntentDimensions> for NameDescription {
+    fn from(value: PaymentIntentDimensions) -> Self {
+        Self {
+            name: value.to_string(),
+            desc: String::new(),
+        }
+    }
+}
+
+#[derive(Debug, serde::Serialize, Eq)]
+pub struct PaymentIntentMetricsBucketIdentifier {
+    pub status: Option<IntentStatus>,
+    pub currency: Option<Currency>,
+    #[serde(rename = "time_range")]
+    pub time_bucket: TimeRange,
+    #[serde(rename = "time_bucket")]
+    #[serde(with = "common_utils::custom_serde::iso8601custom")]
+    pub start_time: time::PrimitiveDateTime,
+}
+
+impl PaymentIntentMetricsBucketIdentifier {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        status: Option<IntentStatus>,
+        currency: Option<Currency>,
+        normalized_time_range: TimeRange,
+    ) -> Self {
+        Self {
+            status,
+            currency,
+            time_bucket: normalized_time_range,
+            start_time: normalized_time_range.start_time,
+        }
+    }
+}
+
+impl Hash for PaymentIntentMetricsBucketIdentifier {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.status.map(|i| i.to_string()).hash(state);
+        self.currency.hash(state);
+        self.time_bucket.hash(state);
+    }
+}
+
+impl PartialEq for PaymentIntentMetricsBucketIdentifier {
+    fn eq(&self, other: &Self) -> bool {
+        let mut left = DefaultHasher::new();
+        self.hash(&mut left);
+        let mut right = DefaultHasher::new();
+        other.hash(&mut right);
+        left.finish() == right.finish()
+    }
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct PaymentIntentMetricsBucketValue {
+    pub successful_smart_retries: Option<u64>,
+    pub total_smart_retries: Option<u64>,
+    pub smart_retried_amount: Option<u64>,
+    pub payment_intent_count: Option<u64>,
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct MetricsBucketResponse {
+    #[serde(flatten)]
+    pub values: PaymentIntentMetricsBucketValue,
+    #[serde(flatten)]
+    pub dimensions: PaymentIntentMetricsBucketIdentifier,
+}
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index bed46f01f19..346fc06f20a 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -38,6 +38,24 @@ use crate::{
 
 impl ApiEventMetric for TimeRange {}
 
+impl ApiEventMetric for GetPaymentIntentFiltersRequest {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::Analytics)
+    }
+}
+
+impl ApiEventMetric for GetPaymentIntentMetricRequest {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::Analytics)
+    }
+}
+
+impl ApiEventMetric for PaymentIntentFiltersResponse {
+    fn get_api_event_type(&self) -> Option<ApiEventsType> {
+        Some(ApiEventsType::Analytics)
+    }
+}
+
 impl_misc_api_event_type!(
     PaymentMethodId,
     PaymentsSessionResponse,
diff --git a/crates/common_utils/src/events.rs b/crates/common_utils/src/events.rs
index 0ef8a6a8bfc..3e3a0da4cab 100644
--- a/crates/common_utils/src/events.rs
+++ b/crates/common_utils/src/events.rs
@@ -65,6 +65,7 @@ pub enum ApiEventsType {
     Poll {
         poll_id: String,
     },
+    Analytics,
 }
 
 impl ApiEventMetric for serde_json::Value {}
diff --git a/crates/router/src/analytics.rs b/crates/router/src/analytics.rs
index 8d8910f7b2c..64f62f48762 100644
--- a/crates/router/src/analytics.rs
+++ b/crates/router/src/analytics.rs
@@ -14,8 +14,9 @@ pub mod routes {
         },
         GenerateReportRequest, GetActivePaymentsMetricRequest, GetApiEventFiltersRequest,
         GetApiEventMetricRequest, GetAuthEventMetricRequest, GetDisputeMetricRequest,
-        GetPaymentFiltersRequest, GetPaymentMetricRequest, GetRefundFilterRequest,
-        GetRefundMetricRequest, GetSdkEventFiltersRequest, GetSdkEventMetricRequest, ReportRequest,
+        GetPaymentFiltersRequest, GetPaymentIntentFiltersRequest, GetPaymentIntentMetricRequest,
+        GetPaymentMetricRequest, GetRefundFilterRequest, GetRefundMetricRequest,
+        GetSdkEventFiltersRequest, GetSdkEventMetricRequest, ReportRequest,
     };
     use error_stack::ResultExt;
 
@@ -37,86 +38,105 @@ pub mod routes {
 
     impl Analytics {
         pub fn server(state: AppState) -> Scope {
-            let mut route = web::scope("/analytics/v1").app_data(web::Data::new(state));
-            {
-                route = route
-                    .service(
-                        web::resource("metrics/payments")
-                            .route(web::post().to(get_payment_metrics)),
-                    )
-                    .service(
-                        web::resource("metrics/refunds").route(web::post().to(get_refunds_metrics)),
-                    )
-                    .service(
-                        web::resource("filters/payments")
-                            .route(web::post().to(get_payment_filters)),
-                    )
-                    .service(
-                        web::resource("filters/refunds").route(web::post().to(get_refund_filters)),
-                    )
-                    .service(web::resource("{domain}/info").route(web::get().to(get_info)))
-                    .service(
-                        web::resource("report/dispute")
-                            .route(web::post().to(generate_dispute_report)),
-                    )
-                    .service(
-                        web::resource("report/refunds")
-                            .route(web::post().to(generate_refund_report)),
-                    )
-                    .service(
-                        web::resource("report/payments")
-                            .route(web::post().to(generate_payment_report)),
-                    )
-                    .service(
-                        web::resource("metrics/sdk_events")
-                            .route(web::post().to(get_sdk_event_metrics)),
-                    )
-                    .service(
-                        web::resource("metrics/active_payments")
-                            .route(web::post().to(get_active_payments_metrics)),
-                    )
-                    .service(
-                        web::resource("filters/sdk_events")
-                            .route(web::post().to(get_sdk_event_filters)),
-                    )
-                    .service(
-                        web::resource("metrics/auth_events")
-                            .route(web::post().to(get_auth_event_metrics)),
-                    )
-                    .service(web::resource("api_event_logs").route(web::get().to(get_api_events)))
-                    .service(web::resource("sdk_event_logs").route(web::post().to(get_sdk_events)))
-                    .service(
-                        web::resource("connector_event_logs")
-                            .route(web::get().to(get_connector_events)),
-                    )
-                    .service(
-                        web::resource("outgoing_webhook_event_logs")
-                            .route(web::get().to(get_outgoing_webhook_events)),
-                    )
-                    .service(
-                        web::resource("filters/api_events")
-                            .route(web::post().to(get_api_event_filters)),
-                    )
-                    .service(
-                        web::resource("metrics/api_events")
-                            .route(web::post().to(get_api_events_metrics)),
-                    )
-                    .service(
-                        web::resource("search").route(web::post().to(get_global_search_results)),
-                    )
-                    .service(
-                        web::resource("search/{domain}").route(web::post().to(get_search_results)),
-                    )
-                    .service(
-                        web::resource("filters/disputes")
-                            .route(web::post().to(get_dispute_filters)),
-                    )
-                    .service(
-                        web::resource("metrics/disputes")
-                            .route(web::post().to(get_dispute_metrics)),
-                    )
-            }
-            route
+            web::scope("/analytics")
+                .app_data(web::Data::new(state))
+                .service(
+                    web::scope("/v1")
+                        .service(
+                            web::resource("metrics/payments")
+                                .route(web::post().to(get_payment_metrics)),
+                        )
+                        .service(
+                            web::resource("metrics/refunds")
+                                .route(web::post().to(get_refunds_metrics)),
+                        )
+                        .service(
+                            web::resource("filters/payments")
+                                .route(web::post().to(get_payment_filters)),
+                        )
+                        .service(
+                            web::resource("filters/refunds")
+                                .route(web::post().to(get_refund_filters)),
+                        )
+                        .service(web::resource("{domain}/info").route(web::get().to(get_info)))
+                        .service(
+                            web::resource("report/dispute")
+                                .route(web::post().to(generate_dispute_report)),
+                        )
+                        .service(
+                            web::resource("report/refunds")
+                                .route(web::post().to(generate_refund_report)),
+                        )
+                        .service(
+                            web::resource("report/payments")
+                                .route(web::post().to(generate_payment_report)),
+                        )
+                        .service(
+                            web::resource("metrics/sdk_events")
+                                .route(web::post().to(get_sdk_event_metrics)),
+                        )
+                        .service(
+                            web::resource("metrics/active_payments")
+                                .route(web::post().to(get_active_payments_metrics)),
+                        )
+                        .service(
+                            web::resource("filters/sdk_events")
+                                .route(web::post().to(get_sdk_event_filters)),
+                        )
+                        .service(
+                            web::resource("metrics/auth_events")
+                                .route(web::post().to(get_auth_event_metrics)),
+                        )
+                        .service(
+                            web::resource("api_event_logs").route(web::get().to(get_api_events)),
+                        )
+                        .service(
+                            web::resource("sdk_event_logs").route(web::post().to(get_sdk_events)),
+                        )
+                        .service(
+                            web::resource("connector_event_logs")
+                                .route(web::get().to(get_connector_events)),
+                        )
+                        .service(
+                            web::resource("outgoing_webhook_event_logs")
+                                .route(web::get().to(get_outgoing_webhook_events)),
+                        )
+                        .service(
+                            web::resource("filters/api_events")
+                                .route(web::post().to(get_api_event_filters)),
+                        )
+                        .service(
+                            web::resource("metrics/api_events")
+                                .route(web::post().to(get_api_events_metrics)),
+                        )
+                        .service(
+                            web::resource("search")
+                                .route(web::post().to(get_global_search_results)),
+                        )
+                        .service(
+                            web::resource("search/{domain}")
+                                .route(web::post().to(get_search_results)),
+                        )
+                        .service(
+                            web::resource("filters/disputes")
+                                .route(web::post().to(get_dispute_filters)),
+                        )
+                        .service(
+                            web::resource("metrics/disputes")
+                                .route(web::post().to(get_dispute_metrics)),
+                        ),
+                )
+                .service(
+                    web::scope("/v2")
+                        .service(
+                            web::resource("/metrics/payments")
+                                .route(web::post().to(get_payment_intents_metrics)),
+                        )
+                        .service(
+                            web::resource("/filters/payments")
+                                .route(web::post().to(get_payment_intents_filters)),
+                        ),
+                )
         }
     }
 
@@ -178,6 +198,42 @@ pub mod routes {
         .await
     }
 
+    /// # Panics
+    ///
+    /// Panics if `json_payload` array does not contain one `GetPaymentIntentMetricRequest` element.
+    pub async fn get_payment_intents_metrics(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Json<[GetPaymentIntentMetricRequest; 1]>,
+    ) -> impl Responder {
+        // safety: This shouldn't panic owing to the data type
+        #[allow(clippy::expect_used)]
+        let payload = json_payload
+            .into_inner()
+            .to_vec()
+            .pop()
+            .expect("Couldn't get GetPaymentIntentMetricRequest");
+        let flow = AnalyticsFlow::GetPaymentIntentMetrics;
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            payload,
+            |state, auth: AuthenticationData, req, _| async move {
+                analytics::payment_intents::get_metrics(
+                    &state.pool,
+                    &auth.merchant_account.merchant_id,
+                    req,
+                )
+                .await
+                .map(ApplicationResponse::Json)
+            },
+            &auth::JWTAuth(Permission::Analytics),
+            api_locking::LockAction::NotApplicable,
+        ))
+        .await
+    }
+
     /// # Panics
     ///
     /// Panics if `json_payload` array does not contain one `GetRefundMetricRequest` element.
@@ -350,6 +406,32 @@ pub mod routes {
         .await
     }
 
+    pub async fn get_payment_intents_filters(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Json<GetPaymentIntentFiltersRequest>,
+    ) -> impl Responder {
+        let flow = AnalyticsFlow::GetPaymentIntentFilters;
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            json_payload.into_inner(),
+            |state, auth: AuthenticationData, req, _| async move {
+                analytics::payment_intents::get_filters(
+                    &state.pool,
+                    req,
+                    &auth.merchant_account.merchant_id,
+                )
+                .await
+                .map(ApplicationResponse::Json)
+            },
+            &auth::JWTAuth(Permission::Analytics),
+            api_locking::LockAction::NotApplicable,
+        ))
+        .await
+    }
+
     pub async fn get_refund_filters(
         state: web::Data<AppState>,
         req: actix_web::HttpRequest,

## 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 -->
Added analytics based on `payment intent`

Currently supporting 4 metrics
- successful smart retries
- total smart retries
- smart retried amount
- payment intent count

filters
- status
- currency

group by
- status
- currency

### 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).
-->
Utilizing payment intents aligns closely with the data needs of merchants, enhancing their ability to manage and process transactions more effectively.
## 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)?
-->
Tested using local data by sending cURL requests through Postman.
Tested using Postgres data from `payment_intents` table and also using Clickhouse data from `payment_intent` table.
<img width="959" alt="Screenshot 2024-06-28 at 12 33 21 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/be420744-297d-4263-aa8d-09653d097fc8">
![Screenshot 2024-06-28 at 12 35 44 PM](https://github.com/juspay/hyperswitch/assets/83278309/878067a4-dd11-47ff-ab76-84196fff78a7)
![Screenshot 2024-06-28 at 12 36 05 PM](https://github.com/juspay/hyperswitch/assets/83278309/4fa8392f-aa18-4117-b88a-e96d9d312ec3)

<img width="562" alt="Screenshot 2024-06-28 at 12 33 38 PM" src="https://github.com/juspay/hyperswitch/assets/83278309/b2e9623c-a341-486c-b6ff-852ed308ecb0">

![Screenshot 2024-06-28 at 12 37 02 PM](https://github.com/juspay/hyperswitch/assets/83278309/b996a927-fb93-45ce-ae44-4a2316368f5f)

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