feat: add identity_key to errors

Author: LuggaPugga

src/batch_queue/mod.rs

@@ -1095,6 +1095,7 @@ mod tests {
                         count: 3,
                         data_entry_id: Uuid::new_v4(),
                         session_id: None,
+                        identity_key: None,
                         build_id: None,
                         created_at: Utc::now(),
                     },

src/handler/collect.rs

@@ -1,7 +1,7 @@
 use super::{
     MODS_EVENT_FIELDS, check_ip_allowed, error_response, extract_known_fields, get_authorization,
     get_client_ip, get_country, insert_error_entries, insert_mods_event, load_project_context,
-    success_response,
+    resolve_identity_key, success_response,
 };
 use crate::batch_queue::{FailedRequest, RequestType, TrackingContext};
 use crate::models::{AppState, Request};
@@ -71,7 +71,7 @@ pub async fn collect(
         id,
         mut data,
         errors,
-        session_id: _,
+        session_id,
     } = req;
 
     let server_id = match id.value().parse::<Uuid>() {
@@ -115,12 +115,21 @@ pub async fn collect(
     }
 
     if let Some(errors) = errors {
-        for error in errors {
+        let fallback_identity = server_id.to_string();
+        for mut error in errors {
+            if error.session_id.is_none() {
+                error.session_id = session_id.clone();
+            }
+            let identity_key = resolve_identity_key(
+                error.session_id.as_deref(),
+                Some(fallback_identity.as_str()),
+            );
             if let Err(e) = insert_error_entries(
                 &state.batch_queue,
                 ctx.project_id,
                 data_entry_id,
                 error,
+                identity_key,
                 Some(tracking_ctx.clone()),
             )
             .await

src/handler/mod.rs

@@ -492,6 +492,7 @@ pub async fn insert_error_entries(
     project_id: Uuid,
     data_entry_id: Uuid,
     data: ErrorTracking,
+    identity_key: Option<String>,
     tracking_ctx: Option<TrackingContext>,
 ) -> Result<(), HandlerResponse> {
     let mut error_rows = Vec::new();
@@ -517,6 +518,7 @@ pub async fn insert_error_entries(
         count: occurrence_count,
         data_entry_id,
         session_id: data.session_id.clone(),
+        identity_key,
         build_id: data.build_id.clone(),
         created_at,
     };
@@ -537,6 +539,20 @@ pub async fn insert_error_entries(
     Ok(())
 }
 
+pub fn resolve_identity_key(
+    session_id: Option<&str>,
+    fallback_identifier: Option<&str>,
+) -> Option<String> {
+    session_id
+        .filter(|value| !value.is_empty())
+        .map(str::to_string)
+        .or_else(|| {
+            fallback_identifier
+                .filter(|value| !value.is_empty())
+                .map(str::to_string)
+        })
+}
+
 pub async fn process_failed_request(
     batch_queue: &BatchQueue,
     pool: &sqlx::PgPool,
@@ -565,7 +581,7 @@ async fn process_collect_request(
         id,
         mut data,
         errors,
-        session_id: _,
+        session_id,
     } = req;
 
     let server_id = id
@@ -598,12 +614,21 @@ async fn process_collect_request(
     if ctx.error_tracking_enabled
         && let Some(errors) = errors
     {
-        for error in errors {
+        let fallback_identity = server_id.to_string();
+        for mut error in errors {
+            if error.session_id.is_none() {
+                error.session_id = session_id.clone();
+            }
+            let identity_key = resolve_identity_key(
+                error.session_id.as_deref(),
+                Some(fallback_identity.as_str()),
+            );
             insert_error_entries(
                 batch_queue,
                 ctx.project_id,
                 data_entry_id,
                 error,
+                identity_key,
                 Some(tracking_ctx.clone()),
             )
             .await
@@ -694,6 +719,7 @@ async fn process_web_request(
         token: token.into(),
         organization_id: ctx.organization_id.as_deref().map(Into::into),
     };
+    let fallback_identity = resolved_user_id.to_string();
 
     let data_entry_id = insert_web_event(
         batch_queue,
@@ -714,11 +740,19 @@ async fn process_web_request(
             if error.session_id.is_none() {
                 error.session_id = parsed.session_id.clone();
             }
+            if error.build_id.is_none() {
+                error.build_id = parsed.build_id.clone();
+            }
+            let identity_key = resolve_identity_key(
+                error.session_id.as_deref(),
+                Some(fallback_identity.as_str()),
+            );
             insert_error_entries(
                 batch_queue,
                 ctx.project_id,
                 data_entry_id,
                 error,
+                identity_key,
                 Some(tracking_ctx.clone()),
             )
             .await
@@ -1102,4 +1136,29 @@ mod tests {
             );
         }
     }
+
+    mod identity_resolution {
+        use super::*;
+
+        #[test]
+        fn prefers_session_id_when_present() {
+            assert_eq!(
+                resolve_identity_key(Some("session-1"), Some("fallback-1")),
+                Some("session-1".to_string())
+            );
+        }
+
+        #[test]
+        fn falls_back_when_session_missing() {
+            assert_eq!(
+                resolve_identity_key(None, Some("fallback-1")),
+                Some("fallback-1".to_string())
+            );
+        }
+
+        #[test]
+        fn ignores_empty_values() {
+            assert_eq!(resolve_identity_key(Some(""), Some("")), None);
+        }
+    }
 }

src/handler/web.rs

@@ -1,8 +1,8 @@
 use super::{
     EncodingQuery, WEB_EVENT_FIELDS, check_ip_allowed, decompress_body, error_response,
     extract_known_fields, get_authorization, get_client_ip, get_country, get_request_origin,
-    insert_error_entries, insert_web_event, load_project_context, success_response,
-    validate_domain,
+    insert_error_entries, insert_web_event, load_project_context, resolve_identity_key,
+    success_response, validate_domain,
 };
 use crate::batch_queue::{FailedRequest, RequestType, TrackingContext};
 use crate::models::{AppState, ErrorTracking};
@@ -178,6 +178,7 @@ pub async fn web(
         token: token.into(),
         organization_id: ctx.organization_id.as_deref().map(Into::into),
     };
+    let fallback_identity = resolved_user_id.to_string();
 
     let data_entry_id = if is_debounced {
         None
@@ -208,11 +209,16 @@ pub async fn web(
             if error.build_id.is_none() {
                 error.build_id = build_id.clone();
             }
+            let identity_key = resolve_identity_key(
+                error.session_id.as_deref(),
+                Some(fallback_identity.as_str()),
+            );
             if let Err(e) = insert_error_entries(
                 &state.batch_queue,
                 ctx.project_id,
                 data_entry_id.unwrap_or_else(Uuid::new_v4),
                 error,
+                identity_key,
                 Some(tracking_ctx.clone()),
             )
             .await

src/tinybird.rs

@@ -81,6 +81,7 @@ pub struct ErrorTrackingRow {
     pub count: u32,
     pub data_entry_id: Uuid,
     pub session_id: Option<String>,
+    pub identity_key: Option<String>,
     pub build_id: Option<String>,
     #[serde(with = "chrono::serde::ts_milliseconds")]
     pub created_at: DateTime<Utc>,