feat: exit events (#53)

Signed-off-by: Henry <mail@henrygressmann.de>

Author: explodingcamera

CHANGELOG.md

@@ -30,6 +30,7 @@ Since this is not a library, this changelog focuses on the changes that are rele
 - Added GeoIP header mappings and presets for Akamai, Cloudflare, CloudFront, Netlify, and Vercel, with MaxMind results taking precedence when available
 - Tracker requests now avoid CORS preflight requests by sending JSON as `text/plain` and the event API accepts tracker JSON regardless of content type
 - Added external authentication with OpenID Connect, Google, and Microsoft Entra ID, including optional user creation and provider access restrictions
+- Improved visit duration accuracy by tracking page exits
 
 ### Security
 

Cargo.lock

@@ -24,7 +24,7 @@ arc-swap = "1.9"
 futures-lite = { version = "2.6", default-features = false, features = ["alloc"] }
 quick_cache = { version = "0.7" }
 tokio = { version = "1.53", default-features = false, features = ["macros", "rt-multi-thread", "signal", "sync"] }
-tokio-util = { version = "0.7", features = ["io"] }
+tokio-util = { version = "0.7", features = ["io", "time"] }
 
 # encoding
 hex = { version = "0.4" }

Cargo.toml

@@ -3,9 +3,9 @@
 <div align="center">
     <h2>
         <img float="left" src="./web/public/favicon.svg" width="16px"/>
-        <a href="https://liwan.dev">liwan.dev</a> - Easy & Privacy-First Web Analytics
+        <a href="https://liwan.dev">liwan.dev</a> - Self-hosted, privacy-first web analytics
     </h2>
-    <div>
+<div>
 
 ![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/explodingcamera/liwan/test.yaml?style=flat-square)
 ![GitHub Release](https://img.shields.io/github/v/release/explodingcamera/liwan?style=flat-square)
@@ -22,18 +22,24 @@
 
 ## Features
 
-- **Quick setup**\
-  Quickly get started with Liwan with a single, self-contained binary . No database or complex setup required. The tracking script is a single line of code that works with any website and less than 1KB in size.
-- **Privacy first**\
-  Liwan respects your users’ privacy by default. No cookies, no cross-site tracking, no persistent identifiers. All data is stored on your server.
-- **Lightweight**\
-  You can run Liwan on a cheap VPS, your old mac mini, or even a Raspberry Pi. Written in Rust and using tokio for async I/O, Liwan is fast and efficient.
-- **Open source**\
-  Fully open source. You can change, extend, and contribute to the codebase.
-- **Accurate data**\
-  Get accurate data about your website’s visitors, page views, referrers, and more. Liwan detects bots and crawlers and filters them out by default.
-- **Real-time analytics**\
-  See your website’s traffic in real-time. Liwan updates the dashboard automatically as new visitors come in.
+**Understand your traffic**\
+See your most-visited pages, where visitors come from, and how traffic changes over time. The dashboard updates automatically, with bot filtering enabled by default.
+
+**Easy to self-host**\
+Run Liwan as a single binary or Docker container. The dashboard and database are built in, with no additional services to manage.
+
+**Privacy first**\
+No tracking cookies or cross-site tracking. Your analytics data stays on your server, and you control what’s collected and how long it’s kept.
+[Read more about data collection](https://liwan.dev/collected-data/).
+
+**Collect only what you need**\
+Choose what data to collect and how long to keep it. Location detail is adjustable, and campaign attribution and session metrics can be disabled independently.
+
+**Lightweight tracking**\
+Add a small tracking script to your website with a single line of HTML. Works with any framework or CMS.
+
+**Single sign-on**\
+Manage accounts with Google Workspaces, Microsoft Entra, or your own OpenID Connect provider like Keycloak or Dex.
 
 ## License
 

README.md

@@ -4,13 +4,17 @@ use anyhow::{Context, Result};
 use arc_swap::ArcSwap;
 use chrono::{DateTime, Local, NaiveTime, TimeZone, Utc};
 use duckdb::{Connection, Result as DuckResult, params};
+use futures_lite::{StreamExt, future};
 use rand::distr::{SampleString, StandardUniform};
 use tokio::sync::mpsc::Receiver;
+use tokio_util::time::DelayQueue;
 
-use crate::app::models::{Event, GeoDetail, ResolvedCollectionSettings, event_params};
+use crate::app::models::{Event, EventExit, GeoDetail, ResolvedCollectionSettings, event_params};
 use crate::app::{DuckDBPool, SqlitePool};
 use crate::utils::duckdb::{ParamVec, repeat_vars};
 
+const EVENT_EXIT_DELAY: std::time::Duration = std::time::Duration::from_secs(30);
+
 #[derive(Clone)]
 pub struct LiwanEvents {
     duckdb: DuckDBPool,
@@ -97,6 +101,41 @@ impl LiwanEvents {
         Ok(())
     }
 
+    /// Process exit updates after a short delay so queued events can reach shared storage first.
+    pub async fn process_exits(&self, mut exits_rx: Receiver<EventExit>) -> Result<()> {
+        let mut pending = DelayQueue::new();
+        let mut channel_closed = false;
+
+        loop {
+            if channel_closed && pending.is_empty() {
+                tracing::info!("Event exit channel closed, stopping event exit processing");
+                return Ok(());
+            }
+
+            tokio::select! {
+                exit = exits_rx.recv(), if !channel_closed => match exit {
+                    Some(exit) => {
+                        pending.insert(exit, EVENT_EXIT_DELAY);
+                    }
+                    None => channel_closed = true,
+                },
+                Some(expired) = pending.next(), if !pending.is_empty() => {
+                    let mut exits = vec![expired.into_inner()];
+                    while let Some(Some(expired)) = future::poll_once(pending.next()).await {
+                        exits.push(expired.into_inner());
+                    }
+
+                    let count = exits.len();
+                    let events = self.clone();
+                    match tokio::task::spawn_blocking(move || events.update_exits(exits)).await? {
+                        Ok(matched) => tracing::debug!(count, matched, "Processed event exits"),
+                        Err(err) => tracing::error!(?err, "Failed to process event exits"),
+                    }
+                }
+            }
+        }
+    }
+
     fn process_events_sync(&self, mut events: Receiver<Event>) -> Result<()> {
         let mut buffer = Vec::with_capacity(1024);
         let conn = self.duckdb.clone();
@@ -139,6 +178,15 @@ impl LiwanEvents {
         }
     }
 
+    fn update_exits(&self, exits: Vec<EventExit>) -> Result<usize> {
+        let conn = self.duckdb.get().context("Failed to get DuckDB connection")?;
+        let mut matched = 0;
+        for exit in exits {
+            matched += usize::from(update_event_exit(&conn, &exit)?);
+        }
+        Ok(matched)
+    }
+
     /// Preview or apply collection-setting pruning for a single entity
     pub fn prune_entity(
         &self,
@@ -205,12 +253,14 @@ impl LiwanEvents {
         }
 
         if !settings.track_sessions {
-            let sql = "entity_id = ? and (time_from_last_event is not null or time_to_next_event is not null)";
+            let sql = "entity_id = ? and (time_from_last_event is not null or time_to_next_event is not null or exited_at is not null)";
             stats.cleared_session_events =
                 count_rows(&conn, &format!("select count(*) from events where {sql}"), params![entity_id])?;
             if !dry_run {
                 conn.execute(
-                    &format!("update events set time_from_last_event = null, time_to_next_event = null where {sql}"),
+                    &format!(
+                        "update events set time_from_last_event = null, time_to_next_event = null, exited_at = null where {sql}"
+                    ),
                     params![entity_id],
                 )?;
             }
@@ -239,6 +289,47 @@ fn count_rows(conn: &Connection, sql: &str, params: impl duckdb::Params) -> Duck
     conn.query_row(sql, params, |row| row.get(0))
 }
 
+fn update_event_exit(conn: &Connection, exit: &EventExit) -> DuckResult<bool> {
+    let sql = "--sql
+        update events
+        set exited_at = greatest(
+            coalesce(exited_at, created_at),
+            least($exited_at::timestamp, created_at + interval '30 minutes')
+        )
+        where
+            entity_id = $entity_id and
+            visitor_group_id = $visitor_group_id and
+            event = $event and
+            fqdn is not distinct from $fqdn and
+            path is not distinct from $path and
+            time_to_next_event is null and
+            created_at = (
+                select max(candidate.created_at)
+                from events candidate
+                where
+                    candidate.entity_id = $entity_id and
+                    candidate.visitor_group_id = $visitor_group_id and
+                    candidate.event = $event and
+                    candidate.fqdn is not distinct from $fqdn and
+                    candidate.path is not distinct from $path and
+                    candidate.time_to_next_event is null and
+                    candidate.created_at <= $exited_at::timestamp and
+                    candidate.created_at >= $exited_at::timestamp - interval '30 minutes'
+            )";
+    let updated = conn.execute(
+        sql,
+        duckdb::named_params! {
+            "entity_id": &exit.entity_id,
+            "visitor_group_id": &exit.visitor_group_id,
+            "event": &exit.event,
+            "fqdn": &exit.fqdn,
+            "path": &exit.path,
+            "exited_at": exit.created_at,
+        },
+    )?;
+    Ok(updated > 0)
+}
+
 fn update_event_times(conn: &Connection, from_time: DateTime<Utc>, entities: &[String]) -> DuckResult<()> {
     if entities.is_empty() {
         return Ok(());
@@ -281,3 +372,71 @@ fn update_event_times(conn: &Connection, from_time: DateTime<Utc>, entities: &[S
     conn.execute(&sql, duckdb::params_from_iter(params))?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::Liwan;
+    use crate::config::Config;
+
+    fn event(created_at: DateTime<Utc>) -> Event {
+        Event {
+            entity_id: "entity-1".to_string(),
+            visitor_group_id: "visitor-1".to_string(),
+            event: "pageview".to_string(),
+            created_at,
+            fqdn: Some("example.com".to_string()),
+            path: Some("/docs".to_string()),
+            referrer: None,
+            platform: None,
+            browser: None,
+            mobile: None,
+            country: None,
+            city: None,
+            utm_source: None,
+            utm_medium: None,
+            utm_campaign: None,
+            utm_content: None,
+            utm_term: None,
+            screen_width: None,
+            orientation: None,
+            track_sessions: true,
+        }
+    }
+
+    #[test]
+    fn exit_updates_only_the_latest_event_without_a_next_event() {
+        let app = Liwan::new_memory(Config::default()).expect("failed to create app");
+        let first = Utc::now() - chrono::Duration::minutes(2);
+        let second = first + chrono::Duration::minutes(1);
+        app.events.append(vec![event(first), event(second)].into_iter()).expect("failed to append events");
+
+        let conn = app.events_conn().expect("failed to get event connection");
+        let matched = update_event_exit(
+            &conn,
+            &EventExit {
+                entity_id: "entity-1".to_string(),
+                visitor_group_id: "visitor-1".to_string(),
+                event: "pageview".to_string(),
+                created_at: second + chrono::Duration::seconds(15),
+                fqdn: Some("example.com".to_string()),
+                path: Some("/docs".to_string()),
+            },
+        )
+        .expect("failed to update exit");
+
+        assert!(matched);
+        let rows = conn
+            .prepare("select exited_at from events order by created_at")
+            .expect("failed to prepare query")
+            .query_map([], |row| row.get::<_, Option<DateTime<Utc>>>(0))
+            .expect("failed to query exits")
+            .collect::<std::result::Result<Vec<_>, _>>()
+            .expect("failed to collect exits");
+        assert_eq!(
+            rows[1].map(|value| value.timestamp_millis()),
+            Some((second + chrono::Duration::seconds(15)).timestamp_millis())
+        );
+        assert_eq!(rows[0], None);
+    }
+}

src/app/core/events.rs

@@ -75,7 +75,8 @@ pub fn dimension_report(
 					visitor_group_id,
 					created_at,
 					time_from_last_event,
-					time_to_next_event
+					time_to_next_event,
+					exited_at
 				from events sd
 				where
 					sd.event = ?::text and

src/app/core/reports/dimension.rs

@@ -167,7 +167,8 @@ pub fn overall_report(
 					e.visitor_group_id,
 					e.created_at,
 					e.time_from_last_event,
-					e.time_to_next_event
+					e.time_to_next_event,
+					e.exited_at
 				from events e
 				where
 					e.event = ?::text and
@@ -181,7 +182,8 @@ pub fn overall_report(
 					sd.visitor_group_id,
 					sd.created_at,
 					sd.time_from_last_event,
-					sd.time_to_next_event
+					sd.time_to_next_event,
+					sd.exited_at
 				from (select * from session_data order by created_at) sd
 				asof join (select * from time_bins order by bin_start) tb
 					on sd.created_at >= tb.bin_start

src/app/core/reports/graph.rs

@@ -110,7 +110,12 @@ pub(super) fn metric_aggregate_sql(metric: Metric, alias: &str) -> String {
         Metric::AvgTimeOnSite => {
             format!(
 				"--sql
-			coalesce(avg(extract(epoch from {alias}.time_to_next_event)) filter (where {alias}.time_to_next_event is not null and {alias}.time_to_next_event <= {SESSION_DURATION_SQL}), 0)"
+			coalesce(avg(extract(epoch from case
+				when {alias}.time_to_next_event is not null and {alias}.time_to_next_event between interval '0 seconds' and {SESSION_DURATION_SQL}
+					then {alias}.time_to_next_event
+				when {alias}.exited_at between {alias}.created_at and {alias}.created_at + {SESSION_DURATION_SQL}
+					then {alias}.exited_at - {alias}.created_at
+			end)), 0)"
 			)
         }
     }

src/app/core/reports/shared.rs

@@ -84,7 +84,8 @@ pub fn overall_stats(
 					e.visitor_group_id,
 					e.created_at,
 					e.time_from_last_event,
-					e.time_to_next_event
+					e.time_to_next_event,
+					e.exited_at
 				from events e
 				where
 					e.event = ?::text and
@@ -113,3 +114,30 @@ pub fn overall_stats(
 
     Ok(result)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::Liwan;
+    use crate::config::Config;
+    use chrono::Duration;
+
+    #[test]
+    fn overall_stats_uses_terminal_event_exit_duration() {
+        let app = Liwan::new_memory(Config::default()).expect("failed to create app");
+        let created_at = Utc::now() - Duration::minutes(5);
+        let exited_at = created_at + Duration::seconds(90);
+        let conn = app.events_conn().expect("failed to get event connection");
+        conn.execute(
+            "insert into events (entity_id, visitor_group_id, event, created_at, fqdn, path, exited_at) values (?, ?, ?, ?, ?, ?, ?)",
+            duckdb::params!["entity-1", "visitor-1", "pageview", created_at, "example.com", "/", exited_at],
+        )
+        .expect("failed to insert event");
+
+        let range = DateRange { start: created_at - Duration::minutes(1), end: exited_at + Duration::minutes(1) };
+        let stats =
+            overall_stats(&conn, &["entity-1".to_string()], "pageview", &range, &[]).expect("failed to build stats");
+
+        assert_eq!(stats.avg_time_on_site, Some(90.0));
+    }
+}

src/app/core/reports/stats.rs

@@ -29,6 +29,17 @@ pub struct Event {
     pub track_sessions: bool,
 }
 
+/// Identifies a stored event that should receive an exit timestamp.
+#[derive(Debug, Clone)]
+pub struct EventExit {
+    pub entity_id: String,
+    pub visitor_group_id: String,
+    pub event: String,
+    pub created_at: DateTime<Utc>,
+    pub fqdn: Option<String>,
+    pub path: Option<String>,
+}
+
 #[derive(Debug, Clone)]
 pub struct Project {
     pub id: String,
@@ -457,6 +468,7 @@ macro_rules! event_params {
             None::<std::time::Duration>,
             $event.screen_width,
             $event.orientation,
+            None::<chrono::DateTime<chrono::Utc>>,
         ]
     };
 }