fix: realtime user analytics (#5129)

config/dashboard.toml

@@ -35,4 +35,5 @@ global_search=true
 dispute_analytics=true
 configure_pmts=false
 branding=false
-totp=false
+totp=false
+live_users_counter=true

crates/analytics/docs/clickhouse/scripts/sdk_events.sql

@@ -210,6 +210,7 @@ CREATE TABLE active_payments (
     `created_at` DateTime64,
     `flow_type` LowCardinality(Nullable(String)),
     INDEX merchantIndex merchant_id TYPE bloom_filter GRANULARITY 1
+    INDEX flowTypeIndex flow_type TYPE bloom_filter GRANULARITY 1
 ) ENGINE = MergeTree
 PARTITION BY toStartOfSecond(created_at)
 ORDER BY 

crates/analytics/src/active_payments/core.rs

@@ -41,6 +41,7 @@ pub async fn get_metrics(
                             &metric_type,
                             &merchant_id_scoped,
                             &publishable_key_scoped,
+                            &req.time_range,
                         )
                         .await
                         .change_context(AnalyticsError::UnknownError);

crates/analytics/src/active_payments/metrics.rs

@@ -1,6 +1,6 @@
 use api_models::analytics::{
     active_payments::{ActivePaymentsMetrics, ActivePaymentsMetricsBucketIdentifier},
-    Granularity,
+    Granularity, TimeRange,
 };
 use time::PrimitiveDateTime;
 
@@ -29,6 +29,7 @@ where
         &self,
         merchant_id: &str,
         publishable_key: &str,
+        time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<
         Vec<(
@@ -52,6 +53,7 @@ where
         &self,
         merchant_id: &str,
         publishable_key: &str,
+        time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<
         Vec<(
@@ -62,7 +64,7 @@ where
         match self {
             Self::ActivePayments => {
                 ActivePayments
-                    .load_metrics(publishable_key, merchant_id, pool)
+                    .load_metrics(publishable_key, merchant_id, time_range, pool)
                     .await
             }
         }

crates/analytics/src/active_payments/metrics/active_payments.rs

@@ -1,11 +1,13 @@
-use api_models::analytics::{active_payments::ActivePaymentsMetricsBucketIdentifier, Granularity};
+use api_models::analytics::{
+    active_payments::ActivePaymentsMetricsBucketIdentifier, Granularity, TimeRange,
+};
 use common_utils::errors::ReportSwitchExt;
 use error_stack::ResultExt;
 use time::PrimitiveDateTime;
 
 use super::ActivePaymentsMetricRow;
 use crate::{
-    query::{Aggregate, FilterTypes, GroupByClause, QueryBuilder, ToSql, Window},
+    query::{Aggregate, FilterTypes, GroupByClause, QueryBuilder, QueryFilter, ToSql, Window},
     types::{AnalyticsCollection, AnalyticsDataSource, MetricsError, MetricsResult},
 };
 
@@ -26,6 +28,7 @@ where
         &self,
         merchant_id: &str,
         publishable_key: &str,
+        time_range: &TimeRange,
         pool: &T,
     ) -> MetricsResult<
         Vec<(
@@ -51,6 +54,23 @@ where
             )
             .switch()?;
 
+        query_builder
+            .add_negative_filter_clause("payment_id", "")
+            .switch()?;
+
+        query_builder
+            .add_custom_filter_clause(
+                "flow_type",
+                "'sdk', 'payment', 'payment_redirection_response'",
+                FilterTypes::In,
+            )
+            .switch()?;
+
+        time_range
+            .set_filter_clause(&mut query_builder)
+            .attach_printable("Error filtering time range")
+            .switch()?;
+
         query_builder
             .execute_query::<ActivePaymentsMetricRow, _>(pool)
             .await

crates/analytics/src/lib.rs

@@ -668,6 +668,7 @@ impl AnalyticsProvider {
         metric: &ActivePaymentsMetrics,
         merchant_id: &str,
         publishable_key: &str,
+        time_range: &TimeRange,
     ) -> types::MetricsResult<
         Vec<(
             ActivePaymentsMetricsBucketIdentifier,
@@ -678,12 +679,12 @@ impl AnalyticsProvider {
             Self::Sqlx(_pool) => Err(report!(MetricsError::NotImplemented)),
             Self::Clickhouse(pool) => {
                 metric
-                    .load_metrics(merchant_id, publishable_key, pool)
+                    .load_metrics(merchant_id, publishable_key, time_range, pool)
                     .await
             }
             Self::CombinedCkh(_sqlx_pool, ckh_pool) | Self::CombinedSqlx(_sqlx_pool, ckh_pool) => {
                 metric
-                    .load_metrics(merchant_id, publishable_key, ckh_pool)
+                    .load_metrics(merchant_id, publishable_key, time_range, ckh_pool)
                     .await
             }
         }