Bug: feat(analytics): add `sessionized_metrics` for refunds (backwards compatibility) and `currency_conversion`



## Current Behaviour

- There is no `sessionized_metrics` module for refunds like there is for `payments` and `payment_intents`
- Update the existing metrics with `currency_conversion` usage.
## Proposed Changes

- create `sessionized_metrics` module for refunds
- implement `currency_conversion` for refunds analytics
diff --git a/crates/analytics/src/payment_intents/core.rs b/crates/analytics/src/payment_intents/core.rs
index 7ea8e9007f7..b242b3dae81 100644
--- a/crates/analytics/src/payment_intents/core.rs
+++ b/crates/analytics/src/payment_intents/core.rs
@@ -69,11 +69,21 @@ pub async fn get_sankey(
                     i.refunds_status.unwrap_or_default().as_ref(),
                     i.attempt_count,
                 ) {
+                    (IntentStatus::Succeeded, SessionizerRefundStatus::FullRefunded, 1) => {
+                        sankey_response.refunded += i.count;
+                        sankey_response.normal_success += i.count
+                    }
+                    (IntentStatus::Succeeded, SessionizerRefundStatus::PartialRefunded, 1) => {
+                        sankey_response.partial_refunded += i.count;
+                        sankey_response.normal_success += i.count
+                    }
                     (IntentStatus::Succeeded, SessionizerRefundStatus::FullRefunded, _) => {
-                        sankey_response.refunded += i.count
+                        sankey_response.refunded += i.count;
+                        sankey_response.smart_retried_success += i.count
                     }
                     (IntentStatus::Succeeded, SessionizerRefundStatus::PartialRefunded, _) => {
-                        sankey_response.partial_refunded += i.count
+                        sankey_response.partial_refunded += i.count;
+                        sankey_response.smart_retried_success += i.count
                     }
                     (
                         IntentStatus::Succeeded
diff --git a/crates/analytics/src/payments/accumulator.rs b/crates/analytics/src/payments/accumulator.rs
index 651eeb0bcfe..5ca9fdf7516 100644
--- a/crates/analytics/src/payments/accumulator.rs
+++ b/crates/analytics/src/payments/accumulator.rs
@@ -387,7 +387,7 @@ impl PaymentMetricsAccumulator {
             payment_processed_count,
             payment_processed_amount_without_smart_retries,
             payment_processed_count_without_smart_retries,
-            payment_processed_amount_usd,
+            payment_processed_amount_in_usd,
             payment_processed_amount_without_smart_retries_usd,
         ) = self.processed_amount.collect();
         let (
@@ -417,7 +417,7 @@ impl PaymentMetricsAccumulator {
             payments_failure_rate_distribution_without_smart_retries,
             failure_reason_count,
             failure_reason_count_without_smart_retries,
-            payment_processed_amount_usd,
+            payment_processed_amount_in_usd,
             payment_processed_amount_without_smart_retries_usd,
         }
     }
diff --git a/crates/analytics/src/payments/core.rs b/crates/analytics/src/payments/core.rs
index bcd009270dc..01a3b1abc15 100644
--- a/crates/analytics/src/payments/core.rs
+++ b/crates/analytics/src/payments/core.rs
@@ -228,7 +228,7 @@ pub async fn get_metrics(
     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 mut total_payment_processed_amount_usd = 0;
+    let mut total_payment_processed_amount_in_usd = 0;
     let mut total_payment_processed_amount_without_smart_retries_usd = 0;
     let query_data: Vec<MetricsBucketResponse> = metrics_accumulator
         .into_iter()
@@ -251,9 +251,9 @@ pub async fn get_metrics(
                     })
                     .map(|amount| (amount * rust_decimal::Decimal::new(100, 0)).to_u64())
                     .unwrap_or_default();
-                collected_values.payment_processed_amount_usd = amount_in_usd;
+                collected_values.payment_processed_amount_in_usd = amount_in_usd;
                 total_payment_processed_amount += amount;
-                total_payment_processed_amount_usd += amount_in_usd.unwrap_or(0);
+                total_payment_processed_amount_in_usd += amount_in_usd.unwrap_or(0);
             }
             if let Some(count) = collected_values.payment_processed_count {
                 total_payment_processed_count += count;
@@ -299,7 +299,7 @@ pub async fn get_metrics(
         query_data,
         meta_data: [PaymentsAnalyticsMetadata {
             total_payment_processed_amount: Some(total_payment_processed_amount),
-            total_payment_processed_amount_usd: Some(total_payment_processed_amount_usd),
+            total_payment_processed_amount_in_usd: Some(total_payment_processed_amount_in_usd),
             total_payment_processed_amount_without_smart_retries: Some(
                 total_payment_processed_amount_without_smart_retries,
             ),
diff --git a/crates/analytics/src/refunds/accumulator.rs b/crates/analytics/src/refunds/accumulator.rs
index 9c51defdcf9..add38c98162 100644
--- a/crates/analytics/src/refunds/accumulator.rs
+++ b/crates/analytics/src/refunds/accumulator.rs
@@ -7,7 +7,7 @@ pub struct RefundMetricsAccumulator {
     pub refund_success_rate: SuccessRateAccumulator,
     pub refund_count: CountAccumulator,
     pub refund_success: CountAccumulator,
-    pub processed_amount: SumAccumulator,
+    pub processed_amount: PaymentProcessedAmountAccumulator,
 }
 
 #[derive(Debug, Default)]
@@ -22,7 +22,7 @@ pub struct CountAccumulator {
 }
 #[derive(Debug, Default)]
 #[repr(transparent)]
-pub struct SumAccumulator {
+pub struct PaymentProcessedAmountAccumulator {
     pub total: Option<i64>,
 }
 
@@ -50,8 +50,8 @@ impl RefundMetricAccumulator for CountAccumulator {
     }
 }
 
-impl RefundMetricAccumulator for SumAccumulator {
-    type MetricOutput = Option<u64>;
+impl RefundMetricAccumulator for PaymentProcessedAmountAccumulator {
+    type MetricOutput = (Option<u64>, Option<u64>);
     #[inline]
     fn add_metrics_bucket(&mut self, metrics: &RefundMetricRow) {
         self.total = match (
@@ -68,7 +68,7 @@ impl RefundMetricAccumulator for SumAccumulator {
     }
     #[inline]
     fn collect(self) -> Self::MetricOutput {
-        self.total.and_then(|i| u64::try_from(i).ok())
+        (self.total.and_then(|i| u64::try_from(i).ok()), Some(0))
     }
 }
 
@@ -98,11 +98,14 @@ impl RefundMetricAccumulator for SuccessRateAccumulator {
 
 impl RefundMetricsAccumulator {
     pub fn collect(self) -> RefundMetricsBucketValue {
+        let (refund_processed_amount, refund_processed_amount_in_usd) =
+            self.processed_amount.collect();
         RefundMetricsBucketValue {
             refund_success_rate: self.refund_success_rate.collect(),
             refund_count: self.refund_count.collect(),
             refund_success_count: self.refund_success.collect(),
-            refund_processed_amount: self.processed_amount.collect(),
+            refund_processed_amount,
+            refund_processed_amount_in_usd,
         }
     }
 }
diff --git a/crates/analytics/src/refunds/core.rs b/crates/analytics/src/refunds/core.rs
index 9c4770c79ee..e3bfa4da9d1 100644
--- a/crates/analytics/src/refunds/core.rs
+++ b/crates/analytics/src/refunds/core.rs
@@ -5,9 +5,12 @@ use api_models::analytics::{
     refunds::{
         RefundDimensions, RefundMetrics, RefundMetricsBucketIdentifier, RefundMetricsBucketResponse,
     },
-    AnalyticsMetadata, GetRefundFilterRequest, GetRefundMetricRequest, MetricsResponse,
-    RefundFilterValue, RefundFiltersResponse,
+    GetRefundFilterRequest, GetRefundMetricRequest, RefundFilterValue, RefundFiltersResponse,
+    RefundsAnalyticsMetadata, RefundsMetricsResponse,
 };
+use bigdecimal::ToPrimitive;
+use common_enums::Currency;
+use currency_conversion::{conversion::convert, types::ExchangeRates};
 use error_stack::ResultExt;
 use router_env::{
     logger,
@@ -29,9 +32,10 @@ use crate::{
 
 pub async fn get_metrics(
     pool: &AnalyticsProvider,
+    ex_rates: &ExchangeRates,
     auth: &AuthInfo,
     req: GetRefundMetricRequest,
-) -> AnalyticsResult<MetricsResponse<RefundMetricsBucketResponse>> {
+) -> AnalyticsResult<RefundsMetricsResponse<RefundMetricsBucketResponse>> {
     let mut metrics_accumulator: HashMap<RefundMetricsBucketIdentifier, RefundMetricsAccumulator> =
         HashMap::new();
     let mut set = tokio::task::JoinSet::new();
@@ -86,16 +90,20 @@ 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 {
-                RefundMetrics::RefundSuccessRate => metrics_builder
-                    .refund_success_rate
-                    .add_metrics_bucket(&value),
-                RefundMetrics::RefundCount => {
+                RefundMetrics::RefundSuccessRate | RefundMetrics::SessionizedRefundSuccessRate => {
+                    metrics_builder
+                        .refund_success_rate
+                        .add_metrics_bucket(&value)
+                }
+                RefundMetrics::RefundCount | RefundMetrics::SessionizedRefundCount => {
                     metrics_builder.refund_count.add_metrics_bucket(&value)
                 }
-                RefundMetrics::RefundSuccessCount => {
+                RefundMetrics::RefundSuccessCount
+                | RefundMetrics::SessionizedRefundSuccessCount => {
                     metrics_builder.refund_success.add_metrics_bucket(&value)
                 }
-                RefundMetrics::RefundProcessedAmount => {
+                RefundMetrics::RefundProcessedAmount
+                | RefundMetrics::SessionizedRefundProcessedAmount => {
                     metrics_builder.processed_amount.add_metrics_bucket(&value)
                 }
             }
@@ -107,18 +115,45 @@ pub async fn get_metrics(
             metrics_accumulator
         );
     }
+    let mut total_refund_processed_amount = 0;
+    let mut total_refund_processed_amount_in_usd = 0;
     let query_data: Vec<RefundMetricsBucketResponse> = metrics_accumulator
         .into_iter()
-        .map(|(id, val)| RefundMetricsBucketResponse {
-            values: val.collect(),
-            dimensions: id,
+        .map(|(id, val)| {
+            let mut collected_values = val.collect();
+            if let Some(amount) = collected_values.refund_processed_amount {
+                let amount_in_usd = id
+                    .currency
+                    .and_then(|currency| {
+                        i64::try_from(amount)
+                            .inspect_err(|e| logger::error!("Amount conversion error: {:?}", e))
+                            .ok()
+                            .and_then(|amount_i64| {
+                                convert(ex_rates, currency, Currency::USD, amount_i64)
+                                    .inspect_err(|e| {
+                                        logger::error!("Currency conversion error: {:?}", e)
+                                    })
+                                    .ok()
+                            })
+                    })
+                    .map(|amount| (amount * rust_decimal::Decimal::new(100, 0)).to_u64())
+                    .unwrap_or_default();
+                collected_values.refund_processed_amount_in_usd = amount_in_usd;
+                total_refund_processed_amount += amount;
+                total_refund_processed_amount_in_usd += amount_in_usd.unwrap_or(0);
+            }
+            RefundMetricsBucketResponse {
+                values: collected_values,
+                dimensions: id,
+            }
         })
         .collect();
 
-    Ok(MetricsResponse {
+    Ok(RefundsMetricsResponse {
         query_data,
-        meta_data: [AnalyticsMetadata {
-            current_time_range: req.time_range,
+        meta_data: [RefundsAnalyticsMetadata {
+            total_refund_processed_amount: Some(total_refund_processed_amount),
+            total_refund_processed_amount_in_usd: Some(total_refund_processed_amount_in_usd),
         }],
     })
 }
diff --git a/crates/analytics/src/refunds/metrics.rs b/crates/analytics/src/refunds/metrics.rs
index 6ecfd8aeb29..c211ea82d7a 100644
--- a/crates/analytics/src/refunds/metrics.rs
+++ b/crates/analytics/src/refunds/metrics.rs
@@ -10,6 +10,7 @@ mod refund_count;
 mod refund_processed_amount;
 mod refund_success_count;
 mod refund_success_rate;
+mod sessionized_metrics;
 use std::collections::HashSet;
 
 use refund_count::RefundCount;
@@ -101,6 +102,26 @@ where
                     .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
                     .await
             }
+            Self::SessionizedRefundSuccessRate => {
+                sessionized_metrics::RefundSuccessRate::default()
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedRefundCount => {
+                sessionized_metrics::RefundCount::default()
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedRefundSuccessCount => {
+                sessionized_metrics::RefundSuccessCount::default()
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
+            Self::SessionizedRefundProcessedAmount => {
+                sessionized_metrics::RefundProcessedAmount::default()
+                    .load_metrics(dimensions, auth, filters, granularity, time_range, pool)
+                    .await
+            }
         }
     }
 }
diff --git a/crates/analytics/src/refunds/metrics/refund_processed_amount.rs b/crates/analytics/src/refunds/metrics/refund_processed_amount.rs
index f0f51a21fe0..6cba5f58fed 100644
--- a/crates/analytics/src/refunds/metrics/refund_processed_amount.rs
+++ b/crates/analytics/src/refunds/metrics/refund_processed_amount.rs
@@ -52,6 +52,7 @@ where
                 alias: Some("total"),
             })
             .switch()?;
+        query_builder.add_select_column("currency").switch()?;
         query_builder
             .add_select_column(Aggregate::Min {
                 field: "created_at",
@@ -78,6 +79,7 @@ where
             query_builder.add_group_by_clause(dim).switch()?;
         }
 
+        query_builder.add_group_by_clause("currency").switch()?;
         if let Some(granularity) = granularity.as_ref() {
             granularity
                 .set_group_by_clause(&mut query_builder)
diff --git a/crates/analytics/src/refunds/metrics/sessionized_metrics/mod.rs b/crates/analytics/src/refunds/metrics/sessionized_metrics/mod.rs
new file mode 100644
index 00000000000..bb404cd3410
--- /dev/null
+++ b/crates/analytics/src/refunds/metrics/sessionized_metrics/mod.rs
@@ -0,0 +1,11 @@
+mod refund_count;
+mod refund_processed_amount;
+mod refund_success_count;
+mod refund_success_rate;
+
+pub(super) use refund_count::RefundCount;
+pub(super) use refund_processed_amount::RefundProcessedAmount;
+pub(super) use refund_success_count::RefundSuccessCount;
+pub(super) use refund_success_rate::RefundSuccessRate;
+
+pub use super::{RefundMetric, RefundMetricAnalytics, RefundMetricRow};
diff --git a/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_count.rs b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_count.rs
new file mode 100644
index 00000000000..c77e1f7a52c
--- /dev/null
+++ b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_count.rs
@@ -0,0 +1,120 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    refunds::{RefundDimensions, RefundFilters, RefundMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::RefundMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct RefundCount {}
+
+#[async_trait::async_trait]
+impl<T> super::RefundMetric<T> for RefundCount
+where
+    T: AnalyticsDataSource + super::RefundMetricAnalytics,
+    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: &[RefundDimensions],
+        auth: &AuthInfo,
+        filters: &RefundFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(RefundMetricsBucketIdentifier, RefundMetricRow)>> {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::RefundSessionized);
+
+        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::<RefundMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    RefundMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        i.refund_status.as_ref().map(|i| i.0.to_string()),
+                        i.connector.clone(),
+                        i.refund_type.as_ref().map(|i| i.0.to_string()),
+                        i.profile_id.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/refunds/metrics/sessionized_metrics/refund_processed_amount.rs b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_processed_amount.rs
new file mode 100644
index 00000000000..c91938228af
--- /dev/null
+++ b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_processed_amount.rs
@@ -0,0 +1,129 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    refunds::{RefundDimensions, RefundFilters, RefundMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::RefundMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+#[derive(Default)]
+pub(crate) struct RefundProcessedAmount {}
+
+#[async_trait::async_trait]
+impl<T> super::RefundMetric<T> for RefundProcessedAmount
+where
+    T: AnalyticsDataSource + super::RefundMetricAnalytics,
+    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: &[RefundDimensions],
+        auth: &AuthInfo,
+        filters: &RefundFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(RefundMetricsBucketIdentifier, RefundMetricRow)>>
+    where
+        T: AnalyticsDataSource + super::RefundMetricAnalytics,
+    {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::RefundSessionized);
+
+        for dim in dimensions.iter() {
+            query_builder.add_select_column(dim).switch()?;
+        }
+
+        query_builder
+            .add_select_column(Aggregate::Sum {
+                field: "refund_amount",
+                alias: Some("total"),
+            })
+            .switch()?;
+        query_builder.add_select_column("currency").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).switch()?;
+        }
+
+        query_builder.add_group_by_clause("currency").switch()?;
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .switch()?;
+        }
+
+        query_builder
+            .add_filter_clause(
+                RefundDimensions::RefundStatus,
+                storage_enums::RefundStatus::Success,
+            )
+            .switch()?;
+
+        query_builder
+            .execute_query::<RefundMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    RefundMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.refund_type.as_ref().map(|i| i.0.to_string()),
+                        i.profile_id.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/refunds/metrics/sessionized_metrics/refund_success_count.rs b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_success_count.rs
new file mode 100644
index 00000000000..332261a3208
--- /dev/null
+++ b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_success_count.rs
@@ -0,0 +1,125 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    refunds::{RefundDimensions, RefundFilters, RefundMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use diesel_models::enums as storage_enums;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::RefundMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(crate) struct RefundSuccessCount {}
+
+#[async_trait::async_trait]
+impl<T> super::RefundMetric<T> for RefundSuccessCount
+where
+    T: AnalyticsDataSource + super::RefundMetricAnalytics,
+    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: &[RefundDimensions],
+        auth: &AuthInfo,
+        filters: &RefundFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(RefundMetricsBucketIdentifier, RefundMetricRow)>>
+    where
+        T: AnalyticsDataSource + super::RefundMetricAnalytics,
+    {
+        let mut query_builder = QueryBuilder::new(AnalyticsCollection::RefundSessionized);
+
+        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).switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder.add_group_by_clause(dim).switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .switch()?;
+        }
+
+        query_builder
+            .add_filter_clause(
+                RefundDimensions::RefundStatus,
+                storage_enums::RefundStatus::Success,
+            )
+            .switch()?;
+        query_builder
+            .execute_query::<RefundMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    RefundMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.refund_type.as_ref().map(|i| i.0.to_string()),
+                        i.profile_id.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<(RefundMetricsBucketIdentifier, RefundMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_success_rate.rs b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_success_rate.rs
new file mode 100644
index 00000000000..35ee0d61b50
--- /dev/null
+++ b/crates/analytics/src/refunds/metrics/sessionized_metrics/refund_success_rate.rs
@@ -0,0 +1,120 @@
+use std::collections::HashSet;
+
+use api_models::analytics::{
+    refunds::{RefundDimensions, RefundFilters, RefundMetricsBucketIdentifier},
+    Granularity, TimeRange,
+};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::RefundMetricRow;
+use crate::{
+    enums::AuthInfo,
+    query::{Aggregate, GroupByClause, QueryBuilder, QueryFilter, SeriesBucket, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+#[derive(Default)]
+pub(crate) struct RefundSuccessRate {}
+
+#[async_trait::async_trait]
+impl<T> super::RefundMetric<T> for RefundSuccessRate
+where
+    T: AnalyticsDataSource + super::RefundMetricAnalytics,
+    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: &[RefundDimensions],
+        auth: &AuthInfo,
+        filters: &RefundFilters,
+        granularity: &Option<Granularity>,
+        time_range: &TimeRange,
+        pool: &T,
+    ) -> MetricsResult<HashSet<(RefundMetricsBucketIdentifier, RefundMetricRow)>>
+    where
+        T: AnalyticsDataSource + super::RefundMetricAnalytics,
+    {
+        let mut query_builder = QueryBuilder::new(AnalyticsCollection::RefundSessionized);
+        let mut dimensions = dimensions.to_vec();
+
+        dimensions.push(RefundDimensions::RefundStatus);
+
+        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).switch()?;
+
+        for dim in dimensions.iter() {
+            query_builder.add_group_by_clause(dim).switch()?;
+        }
+
+        if let Some(granularity) = granularity.as_ref() {
+            granularity
+                .set_group_by_clause(&mut query_builder)
+                .switch()?;
+        }
+
+        query_builder
+            .execute_query::<RefundMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| {
+                Ok((
+                    RefundMetricsBucketIdentifier::new(
+                        i.currency.as_ref().map(|i| i.0),
+                        None,
+                        i.connector.clone(),
+                        i.refund_type.as_ref().map(|i| i.0.to_string()),
+                        i.profile_id.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<(RefundMetricsBucketIdentifier, RefundMetricRow)>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/api_models/src/analytics.rs b/crates/api_models/src/analytics.rs
index 8d63bc3096c..70c0e0e78dc 100644
--- a/crates/api_models/src/analytics.rs
+++ b/crates/api_models/src/analytics.rs
@@ -203,7 +203,7 @@ pub struct AnalyticsMetadata {
 #[derive(Debug, serde::Serialize)]
 pub struct PaymentsAnalyticsMetadata {
     pub total_payment_processed_amount: Option<u64>,
-    pub total_payment_processed_amount_usd: Option<u64>,
+    pub total_payment_processed_amount_in_usd: Option<u64>,
     pub total_payment_processed_amount_without_smart_retries: Option<u64>,
     pub total_payment_processed_amount_without_smart_retries_usd: Option<u64>,
     pub total_payment_processed_count: Option<u64>,
@@ -228,6 +228,11 @@ pub struct PaymentIntentsAnalyticsMetadata {
     pub total_payment_processed_count_without_smart_retries: Option<u64>,
 }
 
+#[derive(Debug, serde::Serialize)]
+pub struct RefundsAnalyticsMetadata {
+    pub total_refund_processed_amount: Option<u64>,
+    pub total_refund_processed_amount_in_usd: Option<u64>,
+}
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GetPaymentFiltersRequest {
@@ -362,6 +367,12 @@ pub struct PaymentIntentsMetricsResponse<T> {
     pub meta_data: [PaymentIntentsAnalyticsMetadata; 1],
 }
 
+#[derive(Debug, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RefundsMetricsResponse<T> {
+    pub query_data: Vec<T>,
+    pub meta_data: [RefundsAnalyticsMetadata; 1],
+}
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct GetApiEventFiltersRequest {
diff --git a/crates/api_models/src/analytics/payments.rs b/crates/api_models/src/analytics/payments.rs
index 1faba79eb37..b34f8c9293c 100644
--- a/crates/api_models/src/analytics/payments.rs
+++ b/crates/api_models/src/analytics/payments.rs
@@ -271,7 +271,7 @@ pub struct PaymentMetricsBucketValue {
     pub payment_count: Option<u64>,
     pub payment_success_count: Option<u64>,
     pub payment_processed_amount: Option<u64>,
-    pub payment_processed_amount_usd: Option<u64>,
+    pub payment_processed_amount_in_usd: Option<u64>,
     pub payment_processed_count: Option<u64>,
     pub payment_processed_amount_without_smart_retries: Option<u64>,
     pub payment_processed_amount_without_smart_retries_usd: Option<u64>,
diff --git a/crates/api_models/src/analytics/refunds.rs b/crates/api_models/src/analytics/refunds.rs
index ef17387d1ea..d981bd4382f 100644
--- a/crates/api_models/src/analytics/refunds.rs
+++ b/crates/api_models/src/analytics/refunds.rs
@@ -88,6 +88,10 @@ pub enum RefundMetrics {
     RefundCount,
     RefundSuccessCount,
     RefundProcessedAmount,
+    SessionizedRefundSuccessRate,
+    SessionizedRefundCount,
+    SessionizedRefundSuccessCount,
+    SessionizedRefundProcessedAmount,
 }
 
 pub mod metric_behaviour {
@@ -176,6 +180,7 @@ pub struct RefundMetricsBucketValue {
     pub refund_count: Option<u64>,
     pub refund_success_count: Option<u64>,
     pub refund_processed_amount: Option<u64>,
+    pub refund_processed_amount_in_usd: Option<u64>,
 }
 #[derive(Debug, serde::Serialize)]
 pub struct RefundMetricsBucketResponse {
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index 7272abffbff..77d8cb117ef 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -168,6 +168,11 @@ impl<T> ApiEventMetric for PaymentIntentsMetricsResponse<T> {
     }
 }
 
+impl<T> ApiEventMetric for RefundsMetricsResponse<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 aba1c2e2efa..fc45e3f80e6 100644
--- a/crates/router/src/analytics.rs
+++ b/crates/router/src/analytics.rs
@@ -652,7 +652,8 @@ pub mod routes {
                     org_id: org_id.clone(),
                     merchant_ids: vec![merchant_id.clone()],
                 };
-                analytics::refunds::get_metrics(&state.pool, &auth, req)
+                let ex_rates = get_forex_exchange_rates(state.clone()).await?;
+                analytics::refunds::get_metrics(&state.pool, &ex_rates, &auth, req)
                     .await
                     .map(ApplicationResponse::Json)
             },
@@ -690,7 +691,8 @@ pub mod routes {
                 let auth: AuthInfo = AuthInfo::OrgLevel {
                     org_id: org_id.clone(),
                 };
-                analytics::refunds::get_metrics(&state.pool, &auth, req)
+                let ex_rates = get_forex_exchange_rates(state.clone()).await?;
+                analytics::refunds::get_metrics(&state.pool, &ex_rates, &auth, req)
                     .await
                     .map(ApplicationResponse::Json)
             },
@@ -735,7 +737,8 @@ pub mod routes {
                     merchant_id: merchant_id.clone(),
                     profile_ids: vec![profile_id.clone()],
                 };
-                analytics::refunds::get_metrics(&state.pool, &auth, req)
+                let ex_rates = get_forex_exchange_rates(state.clone()).await?;
+                analytics::refunds::get_metrics(&state.pool, &ex_rates, &auth, req)
                     .await
                     .map(ApplicationResponse::Json)
             },

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

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

## Description
<!-- Describe your changes in detail -->
- Add sessionized metrics module and implement backwards compatibility.
- Implement currency conversion for refunds analytics

### Minor unrelated changes
- refactor inconsistent naming of fields in some places => [commit](https://github.com/juspay/hyperswitch/pull/6419/commits/08e53d63b5561163712bdfaefebed1caecdb793b)
- fix sankey incorrect formula => [commit](https://github.com/juspay/hyperswitch/pull/6419/commits/6ae2fb68290bf91a6d10178aa620893c3fd45846)
### Fixes #6417 
### 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).
-->
- Refund metrics didn't had a `sessionized_metrics` module and implementation as were there in payments and payment_intents metrics
- Refund metrics weren't implementing currency conversion.
## 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)?
-->
```bash
curl --location 'http://localhost:8080/analytics/v1/org/metrics/refunds' \
--header 'Accept: */*' \
--header 'Accept-Language: en-GB,en-US;q=0.9,en;q=0.8' \
--header 'Connection: keep-alive' \
--header 'Content-Type: application/json' \
--header 'Origin: http://localhost:9000' \
--header 'QueryType: SingleStat' \
--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/130.0.0.0 Safari/537.36' \
--header 'api-key: dev_r3kVZ4URj3fMiHhJJuhHOVqteaXwKyEMGCaFo6PiCS6S9vZsb0ErV4kCVqogM60H' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiMDZlOGRmZWYtMzI0Zi00Yjc0LTg2MTItYzdkZjNhOGZmZTcyIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI3NDM0NTkzIiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTczMTA4ODY3NSwib3JnX2lkIjoib3JnXzl3Y0kxQ2hTOGVEdnRRQmRXak9IIiwicHJvZmlsZV9pZCI6InByb19FNW5lek43YjZUbVB1WlUzbEU1VSJ9.S0OBbdBmFjnRC-1hP-QTVA2fzQESHIvou_8dtmFDCgY' \
--header 'sec-ch-ua: "Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-10-15T18:30:00Z",
            "endTime": "2024-11-07T09:30:00Z"
        },
        "mode": "ORDER",
        "source": "BATCH",
        "metrics": [       
            "sessionized_refund_processed_amount"
        ],
        "delta": true
    }
]'
```
<img width="750" alt="Screenshot 2024-11-07 at 2 56 08 PM" src="https://github.com/user-attachments/assets/6f5ba9b3-1b41-498f-a685-b1e8bf30f78d">

### For sankey changes
```bash 
curl --location 'http://localhost:8080/analytics/v1/profile/metrics/sankey' \
--header 'accept: */*' \
--header 'accept-language: en-US,en;q=0.9' \
--header 'api-key: dev_r3kVZ4URj3fMiHhJJuhHOVqteaXwKyEMGCaFo6PiCS6S9vZsb0ErV4kCVqogM60H' \
--header 'authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoiMDZlOGRmZWYtMzI0Zi00Yjc0LTg2MTItYzdkZjNhOGZmZTcyIiwibWVyY2hhbnRfaWQiOiJtZXJjaGFudF8xNzI3NDM0NTkzIiwicm9sZV9pZCI6Im9yZ19hZG1pbiIsImV4cCI6MTczMTA4ODY3NSwib3JnX2lkIjoib3JnXzl3Y0kxQ2hTOGVEdnRRQmRXak9IIiwicHJvZmlsZV9pZCI6InByb19FNW5lek43YjZUbVB1WlUzbEU1VSJ9.S0OBbdBmFjnRC-1hP-QTVA2fzQESHIvou_8dtmFDCgY' \
--header 'content-type: application/json' \
--header 'origin: https://integ.hyperswitch.io' \
--header 'priority: u=1, i' \
--header 'referer: https://integ.hyperswitch.io/dashboard/new-analytics-payment' \
--header 'sec-ch-ua: "Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"' \
--header 'sec-ch-ua-mobile: ?0' \
--header 'sec-ch-ua-platform: "macOS"' \
--header 'sec-fetch-dest: empty' \
--header 'sec-fetch-mode: cors' \
--header 'sec-fetch-site: same-origin' \
--header 'user-agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36' \
--data '{"startTime":"2024-10-30T18:30:00Z","endTime":"2024-11-08T11:16:00Z"}'
```

<img width="537" alt="Screenshot 2024-11-08 at 3 03 12 PM" src="https://github.com/user-attachments/assets/1fb56fd2-1585-4423-84dd-3883cc3dfc63">

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

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