Bug: Add vector service to accept events from SDK



Need to add a [vector](https://vector.dev/) service to accept events from SDK & insert them to kafka/clickhouse.

This URL can be configured as an env in SDK container/deployment and it can forward the data to the kafka containers in the stack.
diff --git a/config/dashboard.toml b/config/dashboard.toml
index c00b167315a..142cc966500 100644
--- a/config/dashboard.toml
+++ b/config/dashboard.toml
@@ -25,8 +25,8 @@ test_processors=true
 feedback=false
 mixpanel=false
 generate_report=false
-user_journey_analytics=false
-authentication_analytics=false
+user_journey_analytics=true
+authentication_analytics=true
 surcharge=false
 dispute_evidence_upload=false
 paypal_automatic_flow=false
diff --git a/config/vector.yaml b/config/vector.yaml
index d8f5da34666..1511e09a05c 100644
--- a/config/vector.yaml
+++ b/config/vector.yaml
@@ -29,6 +29,11 @@ sources:
   node_metrics:
     type: host_metrics
 
+  sdk_source:
+    type: http_server
+    address: 0.0.0.0:3103
+    encoding: json
+
 transforms:
   plus_1_events:
     type: filter
@@ -56,6 +61,14 @@ transforms:
     source: |-
       .timestamp = from_unix_timestamp!(.created_at, unit: "seconds")
 
+  sdk_transformed:
+    type: throttle
+    inputs:
+      - sdk_source
+    key_field: "{{ .payment_id }}{{ .merchant_id }}"
+    threshold: 1000
+    window_secs: 60
+
 sinks:
   opensearch_events:
     type: elasticsearch
@@ -132,3 +145,16 @@ sinks:
     inputs:
       - vector_metrics
       - node_metrics
+
+  sdk_sink:
+    type: kafka
+    encoding:
+      codec: json
+      except_fields:
+        - "path"
+        - "source_type"
+    inputs:
+      - "sdk_transformed"
+    bootstrap_servers: kafka0:29092
+    topic: hyper-sdk-logs
+    key_field: ".merchant_id"
diff --git a/crates/analytics/docs/clickhouse/scripts/sdk_events.sql b/crates/analytics/docs/clickhouse/scripts/sdk_events.sql
new file mode 100644
index 00000000000..54480fa99b8
--- /dev/null
+++ b/crates/analytics/docs/clickhouse/scripts/sdk_events.sql
@@ -0,0 +1,250 @@
+CREATE TABLE sdk_events_queue ( 
+    `payment_id` Nullable(String),
+    `merchant_id` String,
+    `remote_ip` Nullable(String),
+    `log_type` LowCardinality(Nullable(String)),
+    `event_name` LowCardinality(Nullable(String)),
+    `first_event` LowCardinality(Nullable(String)),
+    `latency` Nullable(UInt32),
+    `timestamp` String,
+    `browser_name` LowCardinality(Nullable(String)),
+    `browser_version` Nullable(String),
+    `platform` LowCardinality(Nullable(String)),
+    `source` LowCardinality(Nullable(String)),
+    `category` LowCardinality(Nullable(String)),
+    `version` LowCardinality(Nullable(String)),
+    `value` Nullable(String),
+    `component` LowCardinality(Nullable(String)),
+    `payment_method` LowCardinality(Nullable(String)),
+    `payment_experience` LowCardinality(Nullable(String))
+) ENGINE = Kafka SETTINGS
+    kafka_broker_list = 'kafka0:29092', 
+    kafka_topic_list = 'hyper-sdk-logs', 
+    kafka_group_name = 'hyper-ckh', 
+    kafka_format = 'JSONEachRow', 
+    kafka_handle_error_mode = 'stream';
+
+CREATE TABLE sdk_events ( 
+    `payment_id` Nullable(String),
+    `merchant_id` String,
+    `remote_ip` Nullable(String),
+    `log_type` LowCardinality(Nullable(String)),
+    `event_name` LowCardinality(Nullable(String)),
+    `first_event` Bool DEFAULT 1,
+    `browser_name` LowCardinality(Nullable(String)),
+    `browser_version` Nullable(String),
+    `platform` LowCardinality(Nullable(String)),
+    `source` LowCardinality(Nullable(String)),
+    `category` LowCardinality(Nullable(String)),
+    `version` LowCardinality(Nullable(String)),
+    `component` LowCardinality(Nullable(String)),
+    `payment_method` LowCardinality(Nullable(String)),
+    `payment_experience` LowCardinality(Nullable(String)) DEFAULT '',
+    `created_at` DateTime DEFAULT now() CODEC(T64, LZ4),
+    `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4),
+    `latency` Nullable(UInt32) DEFAULT 0,
+    `value` Nullable(String),
+    `created_at_precise` DateTime64(3),
+    INDEX paymentMethodIndex payment_method TYPE bloom_filter GRANULARITY 1,
+    INDEX eventIndex event_name TYPE bloom_filter GRANULARITY 1,
+    INDEX platformIndex platform TYPE bloom_filter GRANULARITY 1,
+    INDEX logTypeIndex log_type TYPE bloom_filter GRANULARITY 1,
+    INDEX categoryIndex category TYPE bloom_filter GRANULARITY 1,
+    INDEX sourceIndex source TYPE bloom_filter GRANULARITY 1,
+    INDEX componentIndex component TYPE bloom_filter GRANULARITY 1,
+    INDEX firstEventIndex first_event TYPE bloom_filter GRANULARITY 1
+) ENGINE = MergeTree
+PARTITION BY 
+    toStartOfDay(created_at) 
+ORDER BY 
+    (created_at, merchant_id) 
+TTL 
+    toDateTime(created_at) + toIntervalMonth(6) 
+SETTINGS 
+    index_granularity = 8192
+;
+
+CREATE MATERIALIZED VIEW sdk_events_mv TO sdk_events ( 
+    `payment_id` Nullable(String),
+    `merchant_id` String,
+    `remote_ip` Nullable(String),
+    `log_type` LowCardinality(Nullable(String)),
+    `event_name` LowCardinality(Nullable(String)),
+    `first_event` Bool,
+    `latency` Nullable(UInt32),
+    `browser_name` LowCardinality(Nullable(String)),
+    `browser_version` Nullable(String),
+    `platform` LowCardinality(Nullable(String)),
+    `source` LowCardinality(Nullable(String)),
+    `category` LowCardinality(Nullable(String)),
+    `version` LowCardinality(Nullable(String)),
+    `value` Nullable(String),
+    `component` LowCardinality(Nullable(String)),
+    `payment_method` LowCardinality(Nullable(String)),
+    `payment_experience` LowCardinality(Nullable(String)),
+    `created_at` DateTime64(3),
+    `created_at_precise` DateTime64(3)
+) AS 
+SELECT
+    payment_id,
+    merchant_id,
+    remote_ip,
+    log_type,
+    event_name,
+    multiIf(first_event = 'true', 1, 0) AS first_event,
+    latency,
+    browser_name,
+    browser_version,
+    platform,
+    source,
+    category,
+    version,
+    value,
+    component,
+    payment_method,
+    payment_experience,
+    toDateTime64(timestamp, 3) AS created_at,
+    toDateTime64(timestamp, 3) AS created_at_precise 
+FROM 
+    sdk_events_queue
+WHERE length(_error) = 0;
+
+CREATE TABLE sdk_events_audit (
+    `payment_id` String,
+    `merchant_id` String,
+    `remote_ip` Nullable(String),
+    `log_type` LowCardinality(Nullable(String)),
+    `event_name` LowCardinality(Nullable(String)),
+    `first_event` Bool DEFAULT 1,
+    `browser_name` LowCardinality(Nullable(String)),
+    `browser_version` Nullable(String),
+    `platform` LowCardinality(Nullable(String)),
+    `source` LowCardinality(Nullable(String)),
+    `category` LowCardinality(Nullable(String)),
+    `version` LowCardinality(Nullable(String)),
+    `value` Nullable(String),
+    `component` LowCardinality(Nullable(String)),
+    `payment_method` LowCardinality(Nullable(String)),
+    `payment_experience` LowCardinality(Nullable(String)) DEFAULT '',
+    `created_at` DateTime64(3) DEFAULT now64() CODEC(T64, LZ4),
+    `inserted_at`  DateTime DEFAULT now() CODEC(T64, LZ4), 
+    `latency` Nullable(UInt32) DEFAULT 0
+) ENGINE = MergeTree PARTITION BY merchant_id
+ORDER BY
+    (merchant_id, payment_id)
+    TTL inserted_at + toIntervalMonth(18)
+SETTINGS index_granularity = 8192;
+
+CREATE MATERIALIZED VIEW sdk_events_parse_errors (
+    `topic` String,
+    `partition` Int64,
+    `offset` Int64,
+    `raw` String,
+    `error` String
+) ENGINE = MergeTree
+ORDER BY
+    (topic, partition, offset) SETTINGS index_granularity = 8192 AS
+SELECT
+    _topic AS topic,
+    _partition AS partition,
+    _offset AS offset,
+    _raw_message AS raw,
+    _error AS error
+FROM
+    sdk_events_queue
+WHERE
+    length(_error) > 0;
+
+CREATE MATERIALIZED VIEW sdk_events_audit_mv TO sdk_events_audit (
+    `payment_id` Nullable(String),
+    `merchant_id` String,
+    `remote_ip` Nullable(String),
+    `log_type` LowCardinality(Nullable(String)),
+    `event_name` LowCardinality(Nullable(String)),
+    `first_event` Bool,
+    `latency` Nullable(UInt32),
+    `browser_name` LowCardinality(Nullable(String)),
+    `browser_version` Nullable(String),
+    `platform` LowCardinality(Nullable(String)),
+    `source` LowCardinality(Nullable(String)),
+    `category` LowCardinality(Nullable(String)),
+    `version` LowCardinality(Nullable(String)),
+    `value` Nullable(String),
+    `component` LowCardinality(Nullable(String)),
+    `payment_method` LowCardinality(Nullable(String)),
+    `payment_experience` LowCardinality(Nullable(String)),
+    `created_at` DateTime64(3),
+    `created_at_precise` DateTime64(3),
+    `inserted_at` DateTime DEFAULT now() CODEC(T64, LZ4)
+) AS
+SELECT
+    payment_id,
+    merchant_id,
+    remote_ip,
+    log_type,
+    event_name,
+    multiIf(first_event = 'true', 1, 0) AS first_event,
+    latency,
+    browser_name,
+    browser_version,
+    platform,
+    source,
+    category,
+    version,
+    value,
+    component,
+    payment_method,
+    payment_experience,
+    toDateTime64(timestamp, 3) AS created_at,
+    toDateTime64(timestamp, 3) AS created_at_precise,
+    now() AS inserted_at
+FROM
+    sdk_events_queue
+WHERE
+    (length(_error) = 0)
+    AND (payment_id IS NOT NULL);
+
+CREATE TABLE active_payments ( 
+    `payment_id` Nullable(String),
+    `merchant_id` String,
+    `created_at` DateTime64,
+    `flow_type` LowCardinality(Nullable(String)),
+    INDEX merchantIndex merchant_id TYPE bloom_filter GRANULARITY 1
+) ENGINE = MergeTree
+PARTITION BY toStartOfSecond(created_at)
+ORDER BY 
+    merchant_id
+TTL 
+    toDateTime(created_at) + INTERVAL 60 SECOND
+SETTINGS 
+    index_granularity = 8192;
+
+CREATE MATERIALIZED VIEW sdk_active_payments_mv TO active_payments ( 
+    `payment_id` Nullable(String),
+    `merchant_id` String,
+    `created_at` DateTime64,
+    `flow_type` LowCardinality(Nullable(String))
+) AS 
+SELECT
+    payment_id,
+    merchant_id,
+    toDateTime64(timestamp, 3) AS created_at,
+    'sdk' AS flow_type
+FROM 
+    sdk_events_queue
+WHERE length(_error) = 0;
+
+CREATE MATERIALIZED VIEW api_active_payments_mv TO active_payments ( 
+    `payment_id` Nullable(String),
+    `merchant_id` String,
+    `created_at` DateTime64,
+    `flow_type` LowCardinality(Nullable(String))
+) AS 
+SELECT
+    payment_id,
+    merchant_id,
+    created_at_timestamp AS created_at,
+    flow_type
+FROM 
+    api_events_queue
+WHERE length(_error) = 0;
\ No newline at end of file
diff --git a/crates/analytics/src/active_payments.rs b/crates/analytics/src/active_payments.rs
new file mode 100644
index 00000000000..518e15b7c0d
--- /dev/null
+++ b/crates/analytics/src/active_payments.rs
@@ -0,0 +1,6 @@
+pub mod accumulator;
+mod core;
+pub mod metrics;
+pub use accumulator::{ActivePaymentsMetricAccumulator, ActivePaymentsMetricsAccumulator};
+
+pub use self::core::get_metrics;
diff --git a/crates/analytics/src/active_payments/accumulator.rs b/crates/analytics/src/active_payments/accumulator.rs
new file mode 100644
index 00000000000..2576a8a308b
--- /dev/null
+++ b/crates/analytics/src/active_payments/accumulator.rs
@@ -0,0 +1,47 @@
+use api_models::analytics::active_payments::ActivePaymentsMetricsBucketValue;
+
+use super::metrics::ActivePaymentsMetricRow;
+
+#[derive(Debug, Default)]
+pub struct ActivePaymentsMetricsAccumulator {
+    pub active_payments: CountAccumulator,
+}
+
+#[derive(Debug, Default)]
+#[repr(transparent)]
+pub struct CountAccumulator {
+    pub count: Option<i64>,
+}
+
+pub trait ActivePaymentsMetricAccumulator {
+    type MetricOutput;
+
+    fn add_metrics_bucket(&mut self, metrics: &ActivePaymentsMetricRow);
+
+    fn collect(self) -> Self::MetricOutput;
+}
+
+impl ActivePaymentsMetricAccumulator for CountAccumulator {
+    type MetricOutput = Option<u64>;
+    #[inline]
+    fn add_metrics_bucket(&mut self, metrics: &ActivePaymentsMetricRow) {
+        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 ActivePaymentsMetricsAccumulator {
+    #[allow(dead_code)]
+    pub fn collect(self) -> ActivePaymentsMetricsBucketValue {
+        ActivePaymentsMetricsBucketValue {
+            active_payments: self.active_payments.collect(),
+        }
+    }
+}
diff --git a/crates/analytics/src/active_payments/core.rs b/crates/analytics/src/active_payments/core.rs
new file mode 100644
index 00000000000..0024581bf85
--- /dev/null
+++ b/crates/analytics/src/active_payments/core.rs
@@ -0,0 +1,106 @@
+use std::collections::HashMap;
+
+use api_models::analytics::{
+    active_payments::{
+        ActivePaymentsMetrics, ActivePaymentsMetricsBucketIdentifier, MetricsBucketResponse,
+    },
+    AnalyticsMetadata, GetActivePaymentsMetricRequest, MetricsResponse,
+};
+use error_stack::ResultExt;
+use router_env::{instrument, logger, tracing};
+
+use super::ActivePaymentsMetricsAccumulator;
+use crate::{
+    active_payments::ActivePaymentsMetricAccumulator,
+    errors::{AnalyticsError, AnalyticsResult},
+    AnalyticsProvider,
+};
+
+#[instrument(skip_all)]
+pub async fn get_metrics(
+    pool: &AnalyticsProvider,
+    publishable_key: Option<&String>,
+    merchant_id: Option<&String>,
+    req: GetActivePaymentsMetricRequest,
+) -> AnalyticsResult<MetricsResponse<MetricsBucketResponse>> {
+    let mut metrics_accumulator: HashMap<
+        ActivePaymentsMetricsBucketIdentifier,
+        ActivePaymentsMetricsAccumulator,
+    > = HashMap::new();
+
+    if let Some(publishable_key) = publishable_key {
+        if let Some(merchant_id) = merchant_id {
+            let mut set = tokio::task::JoinSet::new();
+            for metric_type in req.metrics.iter().cloned() {
+                let publishable_key_scoped = publishable_key.to_owned();
+                let merchant_id_scoped = merchant_id.to_owned();
+                let pool = pool.clone();
+                set.spawn(async move {
+                    let data = pool
+                        .get_active_payments_metrics(
+                            &metric_type,
+                            &merchant_id_scoped,
+                            &publishable_key_scoped,
+                        )
+                        .await
+                        .change_context(AnalyticsError::UnknownError);
+                    (metric_type, data)
+                });
+            }
+
+            while let Some((metric, data)) = set
+                .join_next()
+                .await
+                .transpose()
+                .change_context(AnalyticsError::UnknownError)?
+            {
+                logger::info!("Logging metric: {metric} Result: {:?}", data);
+                for (id, value) in data? {
+                    let metrics_builder = metrics_accumulator.entry(id).or_default();
+                    match metric {
+                        ActivePaymentsMetrics::ActivePayments => {
+                            metrics_builder.active_payments.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,
+                }],
+            })
+        } else {
+            logger::error!("Merchant ID not present");
+            Ok(MetricsResponse {
+                query_data: vec![],
+                meta_data: [AnalyticsMetadata {
+                    current_time_range: req.time_range,
+                }],
+            })
+        }
+    } else {
+        logger::error!("Publishable key not present for merchant ID");
+        Ok(MetricsResponse {
+            query_data: vec![],
+            meta_data: [AnalyticsMetadata {
+                current_time_range: req.time_range,
+            }],
+        })
+    }
+}
diff --git a/crates/analytics/src/active_payments/metrics.rs b/crates/analytics/src/active_payments/metrics.rs
new file mode 100644
index 00000000000..43a0b229c8b
--- /dev/null
+++ b/crates/analytics/src/active_payments/metrics.rs
@@ -0,0 +1,70 @@
+use api_models::analytics::{
+    active_payments::{ActivePaymentsMetrics, ActivePaymentsMetricsBucketIdentifier},
+    Granularity,
+};
+use time::PrimitiveDateTime;
+
+use crate::{
+    query::{Aggregate, GroupByClause, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, LoadRow, MetricsResult},
+};
+
+mod active_payments;
+
+use active_payments::ActivePayments;
+
+#[derive(Debug, PartialEq, Eq, serde::Deserialize)]
+pub struct ActivePaymentsMetricRow {
+    pub count: Option<i64>,
+}
+
+pub trait ActivePaymentsMetricAnalytics: LoadRow<ActivePaymentsMetricRow> {}
+
+#[async_trait::async_trait]
+pub trait ActivePaymentsMetric<T>
+where
+    T: AnalyticsDataSource + ActivePaymentsMetricAnalytics,
+{
+    async fn load_metrics(
+        &self,
+        merchant_id: &str,
+        publishable_key: &str,
+        pool: &T,
+    ) -> MetricsResult<
+        Vec<(
+            ActivePaymentsMetricsBucketIdentifier,
+            ActivePaymentsMetricRow,
+        )>,
+    >;
+}
+
+#[async_trait::async_trait]
+impl<T> ActivePaymentsMetric<T> for ActivePaymentsMetrics
+where
+    T: AnalyticsDataSource + ActivePaymentsMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        merchant_id: &str,
+        publishable_key: &str,
+        pool: &T,
+    ) -> MetricsResult<
+        Vec<(
+            ActivePaymentsMetricsBucketIdentifier,
+            ActivePaymentsMetricRow,
+        )>,
+    > {
+        match self {
+            Self::ActivePayments => {
+                ActivePayments
+                    .load_metrics(publishable_key, merchant_id, pool)
+                    .await
+            }
+        }
+    }
+}
diff --git a/crates/analytics/src/active_payments/metrics/active_payments.rs b/crates/analytics/src/active_payments/metrics/active_payments.rs
new file mode 100644
index 00000000000..97b3e69d1ca
--- /dev/null
+++ b/crates/analytics/src/active_payments/metrics/active_payments.rs
@@ -0,0 +1,70 @@
+use api_models::analytics::{active_payments::ActivePaymentsMetricsBucketIdentifier, Granularity};
+use common_utils::errors::ReportSwitchExt;
+use error_stack::ResultExt;
+use time::PrimitiveDateTime;
+
+use super::ActivePaymentsMetricRow;
+use crate::{
+    query::{Aggregate, FilterTypes, GroupByClause, QueryBuilder, ToSql, Window},
+    types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
+};
+
+#[derive(Default)]
+pub(super) struct ActivePayments;
+
+#[async_trait::async_trait]
+impl<T> super::ActivePaymentsMetric<T> for ActivePayments
+where
+    T: AnalyticsDataSource + super::ActivePaymentsMetricAnalytics,
+    PrimitiveDateTime: ToSql<T>,
+    AnalyticsCollection: ToSql<T>,
+    Granularity: GroupByClause<T>,
+    Aggregate<&'static str>: ToSql<T>,
+    Window<&'static str>: ToSql<T>,
+{
+    async fn load_metrics(
+        &self,
+        merchant_id: &str,
+        publishable_key: &str,
+        pool: &T,
+    ) -> MetricsResult<
+        Vec<(
+            ActivePaymentsMetricsBucketIdentifier,
+            ActivePaymentsMetricRow,
+        )>,
+    > {
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::ActivePaymentsAnalytics);
+
+        query_builder
+            .add_select_column(Aggregate::DistinctCount {
+                field: "payment_id",
+                alias: Some("count"),
+            })
+            .switch()?;
+
+        query_builder
+            .add_custom_filter_clause(
+                "merchant_id",
+                format!("'{}','{}'", merchant_id, publishable_key),
+                FilterTypes::In,
+            )
+            .switch()?;
+
+        query_builder
+            .execute_query::<ActivePaymentsMetricRow, _>(pool)
+            .await
+            .change_context(MetricsError::QueryBuildingError)?
+            .change_context(MetricsError::QueryExecutionFailure)?
+            .into_iter()
+            .map(|i| Ok((ActivePaymentsMetricsBucketIdentifier::new(None), i)))
+            .collect::<error_stack::Result<
+                Vec<(
+                    ActivePaymentsMetricsBucketIdentifier,
+                    ActivePaymentsMetricRow,
+                )>,
+                crate::query::PostProcessingError,
+            >>()
+            .change_context(MetricsError::PostProcessingFailure)
+    }
+}
diff --git a/crates/analytics/src/auth_events/metrics/authentication_attempt_count.rs b/crates/analytics/src/auth_events/metrics/authentication_attempt_count.rs
index dd65458b1e3..01110858804 100644
--- a/crates/analytics/src/auth_events/metrics/authentication_attempt_count.rs
+++ b/crates/analytics/src/auth_events/metrics/authentication_attempt_count.rs
@@ -33,7 +33,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(AuthEventMetricsBucketIdentifier, AuthEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
 
         query_builder
             .add_select_column(Aggregate::Count {
diff --git a/crates/analytics/src/auth_events/metrics/authentication_success_count.rs b/crates/analytics/src/auth_events/metrics/authentication_success_count.rs
index 7559abe8e2a..6f0580bb0b5 100644
--- a/crates/analytics/src/auth_events/metrics/authentication_success_count.rs
+++ b/crates/analytics/src/auth_events/metrics/authentication_success_count.rs
@@ -33,7 +33,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(AuthEventMetricsBucketIdentifier, AuthEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
 
         query_builder
             .add_select_column(Aggregate::Count {
diff --git a/crates/analytics/src/auth_events/metrics/challenge_flow_count.rs b/crates/analytics/src/auth_events/metrics/challenge_flow_count.rs
index 51a3b8ed88e..49fef8941f7 100644
--- a/crates/analytics/src/auth_events/metrics/challenge_flow_count.rs
+++ b/crates/analytics/src/auth_events/metrics/challenge_flow_count.rs
@@ -33,7 +33,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(AuthEventMetricsBucketIdentifier, AuthEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
 
         query_builder
             .add_select_column(Aggregate::Count {
diff --git a/crates/analytics/src/auth_events/metrics/frictionless_flow_count.rs b/crates/analytics/src/auth_events/metrics/frictionless_flow_count.rs
index b5760c9dfa2..7ae9d773675 100644
--- a/crates/analytics/src/auth_events/metrics/frictionless_flow_count.rs
+++ b/crates/analytics/src/auth_events/metrics/frictionless_flow_count.rs
@@ -33,7 +33,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(AuthEventMetricsBucketIdentifier, AuthEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
 
         query_builder
             .add_select_column(Aggregate::Count {
diff --git a/crates/analytics/src/auth_events/metrics/three_ds_sdk_count.rs b/crates/analytics/src/auth_events/metrics/three_ds_sdk_count.rs
index be9d7ab0dcf..b08d48b550a 100644
--- a/crates/analytics/src/auth_events/metrics/three_ds_sdk_count.rs
+++ b/crates/analytics/src/auth_events/metrics/three_ds_sdk_count.rs
@@ -33,7 +33,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(AuthEventMetricsBucketIdentifier, AuthEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
 
         query_builder
             .add_select_column(Aggregate::Count {
diff --git a/crates/analytics/src/clickhouse.rs b/crates/analytics/src/clickhouse.rs
index 47294c108f3..ab47397b8af 100644
--- a/crates/analytics/src/clickhouse.rs
+++ b/crates/analytics/src/clickhouse.rs
@@ -7,6 +7,7 @@ use router_env::logger;
 use time::PrimitiveDateTime;
 
 use super::{
+    active_payments::metrics::ActivePaymentsMetricRow,
     auth_events::metrics::AuthEventMetricRow,
     health_check::HealthCheck,
     payments::{
@@ -133,10 +134,12 @@ impl AnalyticsDataSource for ClickhouseClient {
                 TableEngine::CollapsingMergeTree { sign: "sign_flag" }
             }
             AnalyticsCollection::SdkEvents
+            | AnalyticsCollection::SdkEventsAnalytics
             | AnalyticsCollection::ApiEvents
             | AnalyticsCollection::ConnectorEvents
             | AnalyticsCollection::ApiEventsAnalytics
-            | AnalyticsCollection::OutgoingWebhookEvent => TableEngine::BasicTree,
+            | AnalyticsCollection::OutgoingWebhookEvent
+            | AnalyticsCollection::ActivePaymentsAnalytics => TableEngine::BasicTree,
         }
     }
 }
@@ -159,6 +162,7 @@ impl super::refunds::filters::RefundFilterAnalytics for ClickhouseClient {}
 impl super::sdk_events::filters::SdkEventFilterAnalytics for ClickhouseClient {}
 impl super::sdk_events::metrics::SdkEventMetricAnalytics for ClickhouseClient {}
 impl super::sdk_events::events::SdkEventsFilterAnalytics for ClickhouseClient {}
+impl super::active_payments::metrics::ActivePaymentsMetricAnalytics for ClickhouseClient {}
 impl super::auth_events::metrics::AuthEventMetricAnalytics for ClickhouseClient {}
 impl super::api_event::events::ApiLogsFilterAnalytics for ClickhouseClient {}
 impl super::api_event::filters::ApiEventFilterAnalytics for ClickhouseClient {}
@@ -353,6 +357,16 @@ impl TryInto<OutgoingWebhookLogsResult> for serde_json::Value {
     }
 }
 
+impl TryInto<ActivePaymentsMetricRow> for serde_json::Value {
+    type Error = Report<ParsingError>;
+
+    fn try_into(self) -> Result<ActivePaymentsMetricRow, Self::Error> {
+        serde_json::from_value(self).change_context(ParsingError::StructParseFailure(
+            "Failed to parse ActivePaymentsMetricRow in clickhouse results",
+        ))
+    }
+}
+
 impl ToSql<ClickhouseClient> for PrimitiveDateTime {
     fn to_sql(&self, _table_engine: &TableEngine) -> error_stack::Result<String, ParsingError> {
         let format =
@@ -373,12 +387,14 @@ impl ToSql<ClickhouseClient> for AnalyticsCollection {
             Self::Payment => Ok("payment_attempts".to_string()),
             Self::Refund => Ok("refunds".to_string()),
             Self::SdkEvents => Ok("sdk_events_audit".to_string()),
+            Self::SdkEventsAnalytics => Ok("sdk_events".to_string()),
             Self::ApiEvents => Ok("api_events_audit".to_string()),
             Self::ApiEventsAnalytics => Ok("api_events".to_string()),
             Self::PaymentIntent => Ok("payment_intents".to_string()),
             Self::ConnectorEvents => Ok("connector_events_audit".to_string()),
             Self::OutgoingWebhookEvent => Ok("outgoing_webhook_events_audit".to_string()),
             Self::Dispute => Ok("dispute".to_string()),
+            Self::ActivePaymentsAnalytics => Ok("active_payments".to_string()),
         }
     }
 }
@@ -451,6 +467,15 @@ where
                     alias.map_or_else(|| "".to_owned(), |alias| format!(" as {}", alias))
                 )
             }
+            Self::DistinctCount { field, alias } => {
+                format!(
+                    "count(distinct {}){}",
+                    field
+                        .to_sql(table_engine)
+                        .attach_printable("Failed to percentile aggregate")?,
+                    alias.map_or_else(|| "".to_owned(), |alias| format!(" as {}", alias))
+                )
+            }
         })
     }
 }
diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs
index 5c269a4bb18..d3db03a6977 100644
--- a/crates/analytics/src/lib.rs
+++ b/crates/analytics/src/lib.rs
@@ -7,6 +7,7 @@ pub mod payments;
 mod query;
 pub mod refunds;
 
+pub mod active_payments;
 pub mod api_event;
 pub mod auth_events;
 pub mod connector_events;
@@ -32,6 +33,7 @@ pub mod utils;
 use std::sync::Arc;
 
 use api_models::analytics::{
+    active_payments::{ActivePaymentsMetrics, ActivePaymentsMetricsBucketIdentifier},
     api_event::{
         ApiEventDimensions, ApiEventFilters, ApiEventMetrics, ApiEventMetricsBucketIdentifier,
     },
@@ -56,6 +58,7 @@ use storage_impl::config::Database;
 use strum::Display;
 
 use self::{
+    active_payments::metrics::{ActivePaymentsMetric, ActivePaymentsMetricRow},
     auth_events::metrics::{AuthEventMetric, AuthEventMetricRow},
     payments::{
         distribution::{PaymentDistribution, PaymentDistributionRow},
@@ -514,7 +517,7 @@ impl AnalyticsProvider {
         &self,
         metric: &SdkEventMetrics,
         dimensions: &[SdkEventDimensions],
-        pub_key: &str,
+        publishable_key: &str,
         filters: &SdkEventFilters,
         granularity: &Option<Granularity>,
         time_range: &TimeRange,
@@ -523,14 +526,21 @@ impl AnalyticsProvider {
             Self::Sqlx(_pool) => Err(report!(MetricsError::NotImplemented)),
             Self::Clickhouse(pool) => {
                 metric
-                    .load_metrics(dimensions, pub_key, filters, granularity, time_range, pool)
+                    .load_metrics(
+                        dimensions,
+                        publishable_key,
+                        filters,
+                        granularity,
+                        time_range,
+                        pool,
+                    )
                     .await
             }
             Self::CombinedCkh(_sqlx_pool, ckh_pool) | Self::CombinedSqlx(_sqlx_pool, ckh_pool) => {
                 metric
                     .load_metrics(
                         dimensions,
-                        pub_key,
+                        publishable_key,
                         filters,
                         granularity,
                         // Since SDK events are ckh only use ckh here
@@ -542,6 +552,32 @@ impl AnalyticsProvider {
         }
     }
 
+    pub async fn get_active_payments_metrics(
+        &self,
+        metric: &ActivePaymentsMetrics,
+        merchant_id: &str,
+        publishable_key: &str,
+    ) -> types::MetricsResult<
+        Vec<(
+            ActivePaymentsMetricsBucketIdentifier,
+            ActivePaymentsMetricRow,
+        )>,
+    > {
+        match self {
+            Self::Sqlx(_pool) => Err(report!(MetricsError::NotImplemented)),
+            Self::Clickhouse(pool) => {
+                metric
+                    .load_metrics(merchant_id, publishable_key, pool)
+                    .await
+            }
+            Self::CombinedCkh(_sqlx_pool, ckh_pool) | Self::CombinedSqlx(_sqlx_pool, ckh_pool) => {
+                metric
+                    .load_metrics(merchant_id, publishable_key, ckh_pool)
+                    .await
+            }
+        }
+    }
+
     pub async fn get_auth_event_metrics(
         &self,
         metric: &AuthEventMetrics,
@@ -723,6 +759,7 @@ pub enum AnalyticsFlow {
     GetRefundsMetrics,
     GetSdkMetrics,
     GetAuthMetrics,
+    GetActivePaymentsMetrics,
     GetPaymentFilters,
     GetRefundFilters,
     GetSdkEventFilters,
diff --git a/crates/analytics/src/query.rs b/crates/analytics/src/query.rs
index 1ab79f07e43..2fda8fc57cd 100644
--- a/crates/analytics/src/query.rs
+++ b/crates/analytics/src/query.rs
@@ -253,6 +253,10 @@ pub enum Aggregate<R> {
         alias: Option<&'static str>,
         percentile: Option<&'static u8>,
     },
+    DistinctCount {
+        field: R,
+        alias: Option<&'static str>,
+    },
 }
 
 // Window functions in query
diff --git a/crates/analytics/src/sdk_events/filters.rs b/crates/analytics/src/sdk_events/filters.rs
index 9963f51ef94..367db62b183 100644
--- a/crates/analytics/src/sdk_events/filters.rs
+++ b/crates/analytics/src/sdk_events/filters.rs
@@ -24,7 +24,8 @@ where
     Aggregate<&'static str>: ToSql<T>,
     Window<&'static str>: ToSql<T>,
 {
-    let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+    let mut query_builder: QueryBuilder<T> =
+        QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
 
     query_builder.add_select_column(dimension).switch()?;
     time_range
diff --git a/crates/analytics/src/sdk_events/metrics/average_payment_time.rs b/crates/analytics/src/sdk_events/metrics/average_payment_time.rs
index a81a55a2af3..88c96d33efb 100644
--- a/crates/analytics/src/sdk_events/metrics/average_payment_time.rs
+++ b/crates/analytics/src/sdk_events/metrics/average_payment_time.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sdk_events/metrics/load_time.rs b/crates/analytics/src/sdk_events/metrics/load_time.rs
index 07168069b54..4624f304295 100644
--- a/crates/analytics/src/sdk_events/metrics/load_time.rs
+++ b/crates/analytics/src/sdk_events/metrics/load_time.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sdk_events/metrics/payment_attempts.rs b/crates/analytics/src/sdk_events/metrics/payment_attempts.rs
index b2a78188c4f..7cbe93056a9 100644
--- a/crates/analytics/src/sdk_events/metrics/payment_attempts.rs
+++ b/crates/analytics/src/sdk_events/metrics/payment_attempts.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sdk_events/metrics/payment_data_filled_count.rs b/crates/analytics/src/sdk_events/metrics/payment_data_filled_count.rs
index a3c94baeda2..aff78808d1d 100644
--- a/crates/analytics/src/sdk_events/metrics/payment_data_filled_count.rs
+++ b/crates/analytics/src/sdk_events/metrics/payment_data_filled_count.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sdk_events/metrics/payment_method_selected_count.rs b/crates/analytics/src/sdk_events/metrics/payment_method_selected_count.rs
index 11aeac5e6ff..284519441a7 100644
--- a/crates/analytics/src/sdk_events/metrics/payment_method_selected_count.rs
+++ b/crates/analytics/src/sdk_events/metrics/payment_method_selected_count.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sdk_events/metrics/payment_methods_call_count.rs b/crates/analytics/src/sdk_events/metrics/payment_methods_call_count.rs
index 7570f1292e5..e7dc6504036 100644
--- a/crates/analytics/src/sdk_events/metrics/payment_methods_call_count.rs
+++ b/crates/analytics/src/sdk_events/metrics/payment_methods_call_count.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sdk_events/metrics/sdk_initiated_count.rs b/crates/analytics/src/sdk_events/metrics/sdk_initiated_count.rs
index ecbca81acf4..82c2096a2e4 100644
--- a/crates/analytics/src/sdk_events/metrics/sdk_initiated_count.rs
+++ b/crates/analytics/src/sdk_events/metrics/sdk_initiated_count.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sdk_events/metrics/sdk_rendered_count.rs b/crates/analytics/src/sdk_events/metrics/sdk_rendered_count.rs
index ed9e776423a..eba94f0d9ad 100644
--- a/crates/analytics/src/sdk_events/metrics/sdk_rendered_count.rs
+++ b/crates/analytics/src/sdk_events/metrics/sdk_rendered_count.rs
@@ -36,7 +36,8 @@ where
         time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<Vec<(SdkEventMetricsBucketIdentifier, SdkEventMetricRow)>> {
-        let mut query_builder: QueryBuilder<T> = QueryBuilder::new(AnalyticsCollection::SdkEvents);
+        let mut query_builder: QueryBuilder<T> =
+            QueryBuilder::new(AnalyticsCollection::SdkEventsAnalytics);
         let dimensions = dimensions.to_vec();
 
         for dim in dimensions.iter() {
diff --git a/crates/analytics/src/sqlx.rs b/crates/analytics/src/sqlx.rs
index 3a1b7f2d468..76ad9c254be 100644
--- a/crates/analytics/src/sqlx.rs
+++ b/crates/analytics/src/sqlx.rs
@@ -542,6 +542,8 @@ impl ToSql<SqlxClient> for AnalyticsCollection {
             Self::Payment => Ok("payment_attempt".to_string()),
             Self::Refund => Ok("refund".to_string()),
             Self::SdkEvents => Err(error_stack::report!(ParsingError::UnknownError)
+                .attach_printable("SdkEventsAudit table is not implemented for Sqlx"))?,
+            Self::SdkEventsAnalytics => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("SdkEvents table is not implemented for Sqlx"))?,
             Self::ApiEvents => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("ApiEvents table is not implemented for Sqlx"))?,
@@ -550,6 +552,8 @@ impl ToSql<SqlxClient> for AnalyticsCollection {
                 .attach_printable("ConnectorEvents table is not implemented for Sqlx"))?,
             Self::ApiEventsAnalytics => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("ApiEvents table is not implemented for Sqlx"))?,
+            Self::ActivePaymentsAnalytics => Err(error_stack::report!(ParsingError::UnknownError)
+                .attach_printable("ActivePaymentsAnalytics table is not implemented for Sqlx"))?,
             Self::OutgoingWebhookEvent => Err(error_stack::report!(ParsingError::UnknownError)
                 .attach_printable("OutgoingWebhookEvents table is not implemented for Sqlx"))?,
             Self::Dispute => Ok("dispute".to_string()),
@@ -610,6 +614,15 @@ where
                     alias.map_or_else(|| "".to_owned(), |alias| format!(" as {}", alias))
                 )
             }
+            Self::DistinctCount { field, alias } => {
+                format!(
+                    "count(distinct {}){}",
+                    field
+                        .to_sql(table_engine)
+                        .attach_printable("Failed to distinct count aggregate")?,
+                    alias.map_or_else(|| "".to_owned(), |alias| format!(" as {}", alias))
+                )
+            }
         })
     }
 }
diff --git a/crates/analytics/src/types.rs b/crates/analytics/src/types.rs
index 86a9ec86eff..5370fbc25ac 100644
--- a/crates/analytics/src/types.rs
+++ b/crates/analytics/src/types.rs
@@ -26,12 +26,14 @@ pub enum AnalyticsCollection {
     Payment,
     Refund,
     SdkEvents,
+    SdkEventsAnalytics,
     ApiEvents,
     PaymentIntent,
     ConnectorEvents,
     OutgoingWebhookEvent,
     Dispute,
     ApiEventsAnalytics,
+    ActivePaymentsAnalytics,
 }
 
 #[allow(dead_code)]
diff --git a/crates/api_models/src/analytics.rs b/crates/api_models/src/analytics.rs
index d697c8ab271..491420cfc02 100644
--- a/crates/api_models/src/analytics.rs
+++ b/crates/api_models/src/analytics.rs
@@ -4,6 +4,7 @@ use common_utils::pii::EmailStrategy;
 use masking::Secret;
 
 use self::{
+    active_payments::ActivePaymentsMetrics,
     api_event::{ApiEventDimensions, ApiEventMetrics},
     auth_events::AuthEventMetrics,
     disputes::{DisputeDimensions, DisputeMetrics},
@@ -13,6 +14,7 @@ use self::{
 };
 pub use crate::payments::TimeRange;
 
+pub mod active_payments;
 pub mod api_event;
 pub mod auth_events;
 pub mod connector_events;
@@ -151,6 +153,14 @@ pub struct GetAuthEventMetricRequest {
     pub delta: bool,
 }
 
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetActivePaymentsMetricRequest {
+    #[serde(default)]
+    pub metrics: HashSet<ActivePaymentsMetrics>,
+    pub time_range: TimeRange,
+}
+
 #[derive(Debug, serde::Serialize)]
 pub struct AnalyticsMetadata {
     pub current_time_range: TimeRange,
diff --git a/crates/api_models/src/analytics/active_payments.rs b/crates/api_models/src/analytics/active_payments.rs
new file mode 100644
index 00000000000..e7cb2553e01
--- /dev/null
+++ b/crates/api_models/src/analytics/active_payments.rs
@@ -0,0 +1,77 @@
+use std::{
+    collections::hash_map::DefaultHasher,
+    hash::{Hash, Hasher},
+};
+
+use super::NameDescription;
+
+#[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 ActivePaymentsMetrics {
+    ActivePayments,
+}
+
+pub mod metric_behaviour {
+    pub struct ActivePayments;
+}
+
+impl From<ActivePaymentsMetrics> for NameDescription {
+    fn from(value: ActivePaymentsMetrics) -> Self {
+        Self {
+            name: value.to_string(),
+            desc: String::new(),
+        }
+    }
+}
+
+#[derive(Debug, serde::Serialize, Eq)]
+pub struct ActivePaymentsMetricsBucketIdentifier {
+    pub time_bucket: Option<String>,
+}
+
+impl ActivePaymentsMetricsBucketIdentifier {
+    pub fn new(time_bucket: Option<String>) -> Self {
+        Self { time_bucket }
+    }
+}
+
+impl Hash for ActivePaymentsMetricsBucketIdentifier {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.time_bucket.hash(state);
+    }
+}
+
+impl PartialEq for ActivePaymentsMetricsBucketIdentifier {
+    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 ActivePaymentsMetricsBucketValue {
+    pub active_payments: Option<u64>,
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct MetricsBucketResponse {
+    #[serde(flatten)]
+    pub values: ActivePaymentsMetricsBucketValue,
+    #[serde(flatten)]
+    pub dimensions: ActivePaymentsMetricsBucketIdentifier,
+}
diff --git a/crates/api_models/src/events.rs b/crates/api_models/src/events.rs
index 078c27e6db9..bed46f01f19 100644
--- a/crates/api_models/src/events.rs
+++ b/crates/api_models/src/events.rs
@@ -86,6 +86,7 @@ impl_misc_api_event_type!(
     GetInfoResponse,
     GetPaymentMetricRequest,
     GetRefundMetricRequest,
+    GetActivePaymentsMetricRequest,
     GetSdkEventMetricRequest,
     GetAuthEventMetricRequest,
     GetPaymentFiltersRequest,
diff --git a/crates/router/src/analytics.rs b/crates/router/src/analytics.rs
index b471448368f..80b4e188075 100644
--- a/crates/router/src/analytics.rs
+++ b/crates/router/src/analytics.rs
@@ -12,10 +12,10 @@ pub mod routes {
         search::{
             GetGlobalSearchRequest, GetSearchRequest, GetSearchRequestWithIndex, SearchIndex,
         },
-        GenerateReportRequest, GetApiEventFiltersRequest, GetApiEventMetricRequest,
-        GetAuthEventMetricRequest, GetDisputeMetricRequest, GetPaymentFiltersRequest,
-        GetPaymentMetricRequest, GetRefundFilterRequest, GetRefundMetricRequest,
-        GetSdkEventFiltersRequest, GetSdkEventMetricRequest, ReportRequest,
+        GenerateReportRequest, GetActivePaymentsMetricRequest, GetApiEventFiltersRequest,
+        GetApiEventMetricRequest, GetAuthEventMetricRequest, GetDisputeMetricRequest,
+        GetPaymentFiltersRequest, GetPaymentMetricRequest, GetRefundFilterRequest,
+        GetRefundMetricRequest, GetSdkEventFiltersRequest, GetSdkEventMetricRequest, ReportRequest,
     };
     use error_stack::ResultExt;
 
@@ -70,6 +70,10 @@ pub mod routes {
                         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)),
@@ -245,6 +249,43 @@ pub mod routes {
         .await
     }
 
+    /// # Panics
+    ///
+    /// Panics if `json_payload` array does not contain one `GetActivePaymentsMetricRequest` element.
+    pub async fn get_active_payments_metrics(
+        state: web::Data<AppState>,
+        req: actix_web::HttpRequest,
+        json_payload: web::Json<[GetActivePaymentsMetricRequest; 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 GetActivePaymentsMetricRequest");
+        let flow = AnalyticsFlow::GetActivePaymentsMetrics;
+        Box::pin(api::server_wrap(
+            flow,
+            state,
+            &req,
+            payload,
+            |state, auth: AuthenticationData, req, _| async move {
+                analytics::active_payments::get_metrics(
+                    &state.pool,
+                    auth.merchant_account.publishable_key.as_ref(),
+                    Some(&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 `GetAuthEventMetricRequest` element.
diff --git a/docker-compose.yml b/docker-compose.yml
index aff7ab54f45..569ef045064 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -164,8 +164,8 @@ services:
       - HYPERSWITCH_CLIENT_URL=http://localhost:9050
       - SELF_SERVER_URL=http://localhost:5252
       - SDK_ENV=local
-      - ENV_LOGGING_URL=http://localhost:3100
-    labels:  
+      - ENV_LOGGING_URL=http://localhost:3103
+    labels:
       logs: "promtail"
 
   ### Control Center
@@ -405,6 +405,7 @@ services:
     ports:
       - "8686"
       - "9598"
+      - "3103:3103"
     profiles:
       - olap
     environment:

## 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 -->
This PR introduces a real-time analytics feature that allows merchants to track distinct payment IDs on their dashboards. The key components of this feature include a script to create a ClickHouse table with a 1-minute Time-To-Live (TTL) for the data, and an analytics service that counts distinct payment IDs.
Also add the the http server for vector to collect and push logs to kafka in this PR.

## Changes:

**ClickHouse Table Creation Script:**
Added a script to create a ClickHouse table with a 1-minute TTL. This table will store payment events and automatically purge entries older than 1 minute, ensuring that only recent data is kept for real-time analytics.

**Analytics Service:**
Implemented an analytics service that queries the ClickHouse table to count distinct payment IDs. This provides real-time insights into the number of unique payments processed within the last minute.

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->

**1. Start the necessary services using Docker Compose**
```
docker compose up -d kafka-ui clickhouse-server
```
**2. Load the SDK and simulate payments**
**3. Verify Real-Time Analytics**
```
curl --location 'localhost:8080/analytics/v1/metrics/active_payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'Authorization: Bearer {token}' \
--data '[
    {
        "timeRange": {
            "startTime": "2024-06-24T18:30:00Z",
            "endTime": "2024-06-24T18:31:00Z"
        },
        "metrics": [
            "active_payments"
        ]
    }
]'
```
```
{
    "queryData": [
        {
            "active_payments": 2
        }
    ],
    "metaData": [
        {
            "current_time_range": {
              "start_time": "2024-06-24T18:30:00Z",
              "end_time": "2024-06-24T18:31:00Z"
            }
        }
    ]
}
```

## 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`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
