feature: generate duration in favor of session_length aggregation

chat/analytics-server/src/events.rs

@@ -6,7 +6,7 @@ use axum::http::request::Parts;
 use chat_core::User;
 use clickhouse::Row;
 use serde::{Deserialize, Serialize};
-use tracing::info;
+use tracing::{info, warn};
 use uuid::Uuid;
 
 const SESSION_TIMEOUT: i64 = 10 * 60 * 1000; // 10 minutes
@@ -16,6 +16,7 @@ pub struct AnalyticsEventRow {
     // EventContext fields
     pub client_id: String,
     pub session_id: String,
+    pub duration: u32,
     pub app_version: String,
     pub system_os: String,
     pub system_arch: String,
@@ -87,12 +88,19 @@ impl AnalyticsEventRow {
     pub fn set_session_id(&mut self, state: &AppState) {
         if let Some(mut v) = state.sessions.get_mut(&self.client_id) {
             let (session_id, last_server_ts) = v.value_mut();
-            if self.server_ts - *last_server_ts < SESSION_TIMEOUT {
+            let mut duration = self.server_ts - *last_server_ts;
+            if duration < 0 {
+                warn!("Session {} duration is negative, reset to 0", session_id);
+                duration = 0;
+            }
+            if duration < SESSION_TIMEOUT {
                 self.session_id = session_id.clone();
+                self.duration = duration as u32;
                 *last_server_ts = self.server_ts;
             } else {
                 let new_session_id = Uuid::now_v7().to_string();
                 self.session_id = new_session_id.clone();
+                self.duration = 0;
                 info!(
                     "Session {} expired, start a new session: {}",
                     session_id, new_session_id
@@ -103,6 +111,7 @@ impl AnalyticsEventRow {
         } else {
             let session_id = Uuid::now_v7().to_string();
             self.session_id = session_id.clone();
+            self.duration = 0;
             info!("No client id found, start a new session: {}", session_id);
             state
                 .sessions

protos/clickhouse.sql

@@ -2,6 +2,7 @@ CREATE TABLE analytics.analytics_events(
     -- EventContext fields
     client_id String,
     session_id String,
+    duration UInt32,
     app_version String,
     system_os String,
     system_arch String,
@@ -53,8 +54,20 @@ CREATE TABLE analytics.sessions(
     date date,
     client_id String,
     session_id String,
+    app_version String,
+    system_os String,
+    system_arch String,
+    system_locale String,
+    system_timezone String,
+    user_id Nullable(String),
+    ip Nullable(String),
+    user_agent Nullable(String),
+    geo_country Nullable(String),
+    geo_region Nullable(String),
+    geo_city Nullable(String),
     session_start SimpleAggregateFunction(min, DateTime64(3)),
     session_end SimpleAggregateFunction(max, DateTime64(3)),
+    session_length SimpleAggregateFunction(sum, UInt64),
     total_events UInt32) ENGINE = SummingMergeTree()
 ORDER BY
     (
@@ -69,8 +82,20 @@ SELECT
     toDate(server_ts) AS date,
     client_id,
     session_id,
-    minSimpleState(server_ts) AS session_start,
-    maxSimpleState(server_ts) AS session_end,
+    any(app_version) AS app_version,
+    any(system_os) AS system_os,
+    any(system_arch) AS system_arch,
+    any(system_locale) AS system_locale,
+    any(system_timezone) AS system_timezone,
+    any(user_id) AS user_id,
+    any(ip) AS ip,
+    any(user_agent) AS user_agent,
+    any(geo_country) AS geo_country,
+    any(geo_region) AS geo_region,
+    any(geo_city) AS geo_city,
+    min(server_ts) AS session_start,
+    max(server_ts) AS session_end,
+    sum(duration) / 1000 AS session_length,
     count(1) AS total_events
 FROM
     analytics.analytics_events
@@ -83,16 +108,22 @@ GROUP BY
 -- INSERT INTO analytics.sessions...;
 -- query sessions table
 SELECT
-    *,
-    dateDiff('second', session_start, session_end) AS session_length
+    date,
+    client_id,
+    session_id,
+    session_start,
+    session_end,
+    session_length,
+    total_events
 FROM
     analytics.sessions FINAL;
 
 CREATE TABLE analytics.daily_sessions(
     date date,
     client_id String,
     total_session_length SimpleAggregateFunction(sum, UInt64),
-    total_events SimpleAggregateFunction(sum, UInt64)) ENGINE = SummingMergeTree()
+    total_session_events SimpleAggregateFunction(sum, UInt64),
+    unique_users AggregateFunction(uniq, Nullable(String))) ENGINE = SummingMergeTree()
 ORDER BY
     (
         date,
@@ -103,14 +134,27 @@ CREATE MATERIALIZED VIEW analytics.daily_sessions_mv TO analytics.daily_sessions
 SELECT
     date,
     client_id,
-    sumSimpleState(dateDiff('second', session_start, session_end)) AS total_session_length,
-    sumSimpleState(total_events) AS total_events
+    sum(session_length) AS total_session_length,
+    sum(total_events) AS total_session_events,
+    uniqState(user_id) AS unique_users
 FROM
     analytics.sessions
 GROUP BY
     date,
     client_id;
 
+SELECT
+    date,
+    client_id,
+    sum(total_session_length) AS total_session_length,
+    sum(total_session_events) AS total_session_events,
+    uniqMerge(unique_users) AS unique_users
+FROM
+    analytics.daily_sessions
+GROUP BY
+    date,
+    client_id;
+
 -- Insert sample data for AppStartEvent
 INSERT INTO analytics.analytics_events(client_id, session_id, app_version, system_os, system_arch, system_locale, system_timezone, client_ts, server_ts, event_type)
     VALUES ('client_001', 'session_001', '1.0.0', 'macOS', 'x86_64', 'en-US', 'America/New_York', now(), now(), 'AppStart');