Single-binary self-hosted website analytics on Rust axum: collector API, dashboards, world map, and PDF reports.
analyticsaxumdockerrustself-hostedsqliteviteweb-analytics
1//! Dashboard aggregation queries. Mirror of `properties/queries.py` from the
2//! Django version, but talking to the hot-field schema so most aggregations
3//! become straight COUNT(*) over typed columns.
4//!
5//! Time arithmetic uses unix milliseconds (matches `events.created_at`).
6
7use chrono::{DateTime, Duration, NaiveDate, TimeZone, Utc};
8use serde::Serialize;
9use sqlx::SqlitePool;
10use uuid::Uuid;
11
12use crate::models::CustomCard;
13
14const BUILT_IN_EVENTS: &[&str] = &["session_start", "page_view", "page_leave", "click", "scroll"];
15
16const TIME_ON_PAGE_MIN_S: f64 = 1.0;
17const TIME_ON_PAGE_MAX_S: f64 = 30.0 * 60.0;
18
19#[derive(Debug, Clone, Serialize)]
20pub struct LabelCount {
21 pub label: String,
22 pub count: i64,
23}
24
25#[derive(Debug, Clone, Serialize)]
26pub struct GraphPoint {
27 pub label: String,
28 pub count: i64,
29}
30
31#[derive(Debug, Clone, Serialize)]
32pub struct EventCard {
33 pub name: String,
34 pub value: serde_json::Value,
35 pub percent_change: i64,
36 #[serde(skip_serializing_if = "Option::is_none")]
37 pub help_text: Option<String>,
38}
39
40#[derive(Debug, Clone, Serialize)]
41pub struct CustomEventDescriptor {
42 pub event: String,
43 pub active: bool,
44}
45
46#[derive(Debug, Clone, Serialize, Default)]
47pub struct BotTraffic {
48 pub total: i64,
49 pub top_bots: Vec<LabelCount>,
50 pub top_pages: Vec<LabelCount>,
51}
52
53#[derive(Debug, Clone, Default)]
54pub struct EventCounts {
55 pub session_start: i64,
56 pub page_view: i64,
57 pub click: i64,
58 pub scroll: i64,
59 pub total: i64,
60}
61
62fn pct_change(current: f64, previous: f64) -> i64 {
63 if previous == 0.0 {
64 return 0;
65 }
66 ((current - previous) / previous * 100.0).round() as i64
67}
68
69fn filter_clause(filter_url: Option<&str>) -> (&'static str, Option<String>) {
70 match filter_url {
71 Some(_) => (" AND url = ?", filter_url.map(|s| s.to_string())),
72 None => ("", None),
73 }
74}
75
76/// Total unique user_ids seen in the last 30 minutes.
77pub async fn total_live_users(pool: &SqlitePool, property_id: &Uuid) -> i64 {
78 let cutoff = (Utc::now() - Duration::minutes(30)).timestamp_millis();
79 sqlx::query_scalar::<_, i64>(
80 "SELECT COUNT(DISTINCT user_id) FROM events \
81 WHERE property_id = ? AND created_at >= ? AND user_id IS NOT NULL",
82 )
83 .bind(property_id.as_bytes().to_vec())
84 .bind(cutoff)
85 .fetch_one(pool)
86 .await
87 .unwrap_or(0)
88}
89
90pub async fn event_counts(
91 pool: &SqlitePool,
92 property_id: &Uuid,
93 start_ms: i64,
94 end_ms: i64,
95 filter_url: Option<&str>,
96) -> EventCounts {
97 let (extra_sql, extra_bind) = filter_clause(filter_url);
98 let sql = format!(
99 "SELECT \
100 SUM(CASE WHEN event = 'session_start' THEN 1 ELSE 0 END) AS session_start, \
101 SUM(CASE WHEN event = 'page_view' THEN 1 ELSE 0 END) AS page_view, \
102 SUM(CASE WHEN event = 'click' THEN 1 ELSE 0 END) AS click, \
103 SUM(CASE WHEN event = 'scroll' THEN 1 ELSE 0 END) AS scroll, \
104 COUNT(*) AS total \
105 FROM events \
106 WHERE property_id = ? AND created_at >= ? AND created_at <= ?{}",
107 extra_sql
108 );
109 let mut q = sqlx::query_as::<_, (Option<i64>, Option<i64>, Option<i64>, Option<i64>, i64)>(&sql)
110 .bind(property_id.as_bytes().to_vec())
111 .bind(start_ms)
112 .bind(end_ms);
113 if let Some(v) = extra_bind {
114 q = q.bind(v);
115 }
116 let row = q.fetch_one(pool).await.unwrap_or((None, None, None, None, 0));
117 EventCounts {
118 session_start: row.0.unwrap_or(0),
119 page_view: row.1.unwrap_or(0),
120 click: row.2.unwrap_or(0),
121 scroll: row.3.unwrap_or(0),
122 total: row.4,
123 }
124}
125
126async fn engaged_users(
127 pool: &SqlitePool,
128 property_id: &Uuid,
129 start_ms: i64,
130 end_ms: i64,
131 filter_url: Option<&str>,
132 session_starts: i64,
133) -> f64 {
134 if session_starts == 0 {
135 return 0.0;
136 }
137 let (extra_sql, extra_bind) = filter_clause(filter_url);
138 let sql = format!(
139 "SELECT COUNT(*) FROM ( \
140 SELECT user_id FROM events \
141 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
142 AND user_id IS NOT NULL{} \
143 GROUP BY user_id HAVING COUNT(*) >= 10 \
144 )",
145 extra_sql
146 );
147 let mut q = sqlx::query_scalar::<_, i64>(&sql)
148 .bind(property_id.as_bytes().to_vec())
149 .bind(start_ms)
150 .bind(end_ms);
151 if let Some(v) = extra_bind {
152 q = q.bind(v);
153 }
154 let engaged = q.fetch_one(pool).await.unwrap_or(0);
155 ((engaged as f64) / (session_starts as f64) * 100.0 * 100.0).round() / 100.0
156}
157
158async fn avg_time_on_page(
159 pool: &SqlitePool,
160 property_id: &Uuid,
161 start_ms: i64,
162 end_ms: i64,
163 filter_url: Option<&str>,
164) -> f64 {
165 let (extra_sql, extra_bind) = filter_clause(filter_url);
166 let sql = format!(
167 "SELECT AVG(time_on_page_ms / 1000.0) FROM events \
168 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
169 AND event = 'page_leave' \
170 AND time_on_page_ms IS NOT NULL \
171 AND time_on_page_ms / 1000.0 BETWEEN ? AND ?{}",
172 extra_sql
173 );
174 let mut q = sqlx::query_scalar::<_, Option<f64>>(&sql)
175 .bind(property_id.as_bytes().to_vec())
176 .bind(start_ms)
177 .bind(end_ms)
178 .bind(TIME_ON_PAGE_MIN_S)
179 .bind(TIME_ON_PAGE_MAX_S);
180 if let Some(v) = extra_bind {
181 q = q.bind(v);
182 }
183 let avg = q.fetch_one(pool).await.unwrap_or(None).unwrap_or(0.0);
184 (avg * 100.0).round() / 100.0
185}
186
187pub async fn standard_event_cards(
188 pool: &SqlitePool,
189 property_id: &Uuid,
190 start_ms: i64,
191 end_ms: i64,
192 prev_start_ms: i64,
193 prev_end_ms: i64,
194 filter_url: Option<&str>,
195) -> Vec<EventCard> {
196 let cur = event_counts(pool, property_id, start_ms, end_ms, filter_url).await;
197 let prev = event_counts(pool, property_id, prev_start_ms, prev_end_ms, filter_url).await;
198
199 let mut cards = vec![
200 EventCard {
201 name: "Total session starts".into(),
202 value: cur.session_start.into(),
203 percent_change: pct_change(cur.session_start as f64, prev.session_start as f64),
204 help_text: Some("Unique users visiting your site for your selected date range.".into()),
205 },
206 EventCard {
207 name: "Total page views".into(),
208 value: cur.page_view.into(),
209 percent_change: pct_change(cur.page_view as f64, prev.page_view as f64),
210 help_text: Some("Total pages viewed for your selected date range.".into()),
211 },
212 EventCard {
213 name: "Total clicks".into(),
214 value: cur.click.into(),
215 percent_change: pct_change(cur.click as f64, prev.click as f64),
216 help_text: Some("Total clicks users made on all your pages for your selected date range.".into()),
217 },
218 EventCard {
219 name: "Total scrolls".into(),
220 value: cur.scroll.into(),
221 percent_change: pct_change(cur.scroll as f64, prev.scroll as f64),
222 help_text: Some("Total scrolls users made on all your pages for your selected date range.".into()),
223 },
224 EventCard {
225 name: "Total events".into(),
226 value: cur.total.into(),
227 percent_change: pct_change(cur.total as f64, prev.total as f64),
228 help_text: Some("All events for your selected date range, including custom events.".into()),
229 },
230 ];
231
232 let eng_cur =
233 engaged_users(pool, property_id, start_ms, end_ms, filter_url, cur.session_start).await;
234 let eng_prev = engaged_users(
235 pool,
236 property_id,
237 prev_start_ms,
238 prev_end_ms,
239 filter_url,
240 prev.session_start,
241 )
242 .await;
243 cards.push(EventCard {
244 name: "Total user engagement".into(),
245 value: format!("{eng_cur}%").into(),
246 percent_change: pct_change(eng_cur, eng_prev),
247 help_text: Some("An engaged user is a user with more than 10 events collected for your selected date range.".into()),
248 });
249
250 let t_cur = avg_time_on_page(pool, property_id, start_ms, end_ms, filter_url).await;
251 let t_prev = avg_time_on_page(pool, property_id, prev_start_ms, prev_end_ms, filter_url).await;
252 cards.push(EventCard {
253 name: "Avg. time on page".into(),
254 value: format!("{t_cur}s").into(),
255 percent_change: pct_change(t_cur, t_prev),
256 help_text: Some("Average time a user spends on each page. Sessions over 30 minutes are excluded as idle.".into()),
257 });
258
259 cards
260}
261
262pub async fn custom_event_cards(
263 pool: &SqlitePool,
264 property_id: &Uuid,
265 custom_cards: &[CustomCard],
266 start_ms: i64,
267 end_ms: i64,
268 prev_start_ms: i64,
269 prev_end_ms: i64,
270 filter_url: Option<&str>,
271) -> (Vec<EventCard>, Vec<CustomEventDescriptor>) {
272 // All non-built-in event names that have ever been seen for this property.
273 let placeholders = std::iter::repeat("?")
274 .take(BUILT_IN_EVENTS.len())
275 .collect::<Vec<_>>()
276 .join(",");
277 let sql = format!(
278 "SELECT DISTINCT event FROM events \
279 WHERE property_id = ? AND event NOT IN ({}) \
280 ORDER BY event",
281 placeholders
282 );
283 let mut q = sqlx::query_scalar::<_, String>(&sql).bind(property_id.as_bytes().to_vec());
284 for built in BUILT_IN_EVENTS {
285 q = q.bind(built);
286 }
287 let names: Vec<String> = q.fetch_all(pool).await.unwrap_or_default();
288
289 let active: std::collections::HashSet<&str> = custom_cards
290 .iter()
291 .filter(|c| c.value)
292 .map(|c| c.event.as_str())
293 .collect();
294
295 let descriptors: Vec<CustomEventDescriptor> = names
296 .iter()
297 .map(|n| CustomEventDescriptor {
298 event: n.clone(),
299 active: active.contains(n.as_str()),
300 })
301 .collect();
302
303 if active.is_empty() {
304 return (Vec::new(), descriptors);
305 }
306
307 // Aggregate counts for active custom events in current and previous periods.
308 let active_names: Vec<&str> = active.iter().copied().collect();
309 let count_for = |period_start: i64, period_end: i64| {
310 let placeholders = std::iter::repeat("?")
311 .take(active_names.len())
312 .collect::<Vec<_>>()
313 .join(",");
314 let (extra_sql, extra_bind) = filter_clause(filter_url);
315 let sql = format!(
316 "SELECT event, COUNT(*) FROM events \
317 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
318 AND event IN ({}){} \
319 GROUP BY event",
320 placeholders, extra_sql
321 );
322 let pool = pool.clone();
323 let property_id = *property_id;
324 let names: Vec<String> = active_names.iter().map(|s| (*s).to_string()).collect();
325 let extra_bind = extra_bind.clone();
326 async move {
327 let mut q = sqlx::query_as::<_, (String, i64)>(&sql)
328 .bind(property_id.as_bytes().to_vec())
329 .bind(period_start)
330 .bind(period_end);
331 for n in &names {
332 q = q.bind(n);
333 }
334 if let Some(v) = extra_bind {
335 q = q.bind(v);
336 }
337 q.fetch_all(&pool).await.unwrap_or_default()
338 }
339 };
340
341 let cur_rows = count_for(start_ms, end_ms).await;
342 let prev_rows = count_for(prev_start_ms, prev_end_ms).await;
343 let cur_map: std::collections::HashMap<String, i64> = cur_rows.into_iter().collect();
344 let prev_map: std::collections::HashMap<String, i64> = prev_rows.into_iter().collect();
345
346 let cards: Vec<EventCard> = active_names
347 .iter()
348 .map(|name| {
349 let v = *cur_map.get(*name).unwrap_or(&0);
350 let p = *prev_map.get(*name).unwrap_or(&0);
351 EventCard {
352 name: (*name).to_string(),
353 value: v.into(),
354 percent_change: pct_change(v as f64, p as f64),
355 help_text: None,
356 }
357 })
358 .collect();
359
360 (cards, descriptors)
361}
362
363/// Time-series chart data. Buckets daily/weekly/monthly based on the date range,
364/// stepping backwards from `end_date`.
365pub async fn events_graph(
366 pool: &SqlitePool,
367 property_id: &Uuid,
368 start_ms: i64,
369 end_ms: i64,
370 filter_url: Option<&str>,
371 end_date: NaiveDate,
372 range_days: i64,
373) -> Vec<GraphPoint> {
374 let (extra_sql, extra_bind) = filter_clause(filter_url);
375 let sql = format!(
376 "SELECT date(created_at / 1000, 'unixepoch') AS day, COUNT(*) \
377 FROM events \
378 WHERE property_id = ? AND created_at >= ? AND created_at <= ?{} \
379 GROUP BY day",
380 extra_sql
381 );
382 let mut q = sqlx::query_as::<_, (String, i64)>(&sql)
383 .bind(property_id.as_bytes().to_vec())
384 .bind(start_ms)
385 .bind(end_ms);
386 if let Some(v) = extra_bind {
387 q = q.bind(v);
388 }
389 let rows = q.fetch_all(pool).await.unwrap_or_default();
390
391 let mut by_day: std::collections::HashMap<NaiveDate, i64> =
392 std::collections::HashMap::with_capacity(rows.len());
393 for (s, c) in rows {
394 if let Ok(d) = NaiveDate::parse_from_str(&s, "%Y-%m-%d") {
395 by_day.insert(d, c);
396 }
397 }
398
399 let bucket_sum = |start: NaiveDate, days: i64| -> i64 {
400 (0..days)
401 .map(|j| {
402 start
403 .checked_add_signed(Duration::days(j))
404 .and_then(|d| by_day.get(&d).copied())
405 .unwrap_or(0)
406 })
407 .sum()
408 };
409
410 let mut points = Vec::new();
411 if range_days <= 28 {
412 for i in 0..range_days {
413 if let Some(d) = end_date.checked_sub_signed(Duration::days(i)) {
414 points.push((d, by_day.get(&d).copied().unwrap_or(0)));
415 }
416 }
417 } else if range_days <= 6 * 28 {
418 let weeks = range_days / 7;
419 for w in 0..weeks {
420 if let Some(d) = end_date.checked_sub_signed(Duration::days(7 * w)) {
421 points.push((d, bucket_sum(d, 7)));
422 }
423 }
424 } else {
425 let months = range_days / 28;
426 for m in 0..months {
427 if let Some(d) = end_date.checked_sub_signed(Duration::days(28 * m)) {
428 points.push((d, bucket_sum(d, 28)));
429 }
430 }
431 }
432 points.sort_by_key(|p| p.0);
433 points
434 .into_iter()
435 .map(|(d, c)| GraphPoint { label: d.format("%b %-d").to_string(), count: c })
436 .collect()
437}
438
439async fn top_by_column(
440 pool: &SqlitePool,
441 property_id: &Uuid,
442 start_ms: i64,
443 end_ms: i64,
444 filter_url: Option<&str>,
445 column: &str,
446 event: Option<&str>,
447 limit: i64,
448 distinct_users: bool,
449) -> Vec<LabelCount> {
450 // For user-property breakdowns (device/browser/platform), count one row
451 // per anonymous user_id. For everything else, count raw events.
452 let count_expr = if distinct_users {
453 "COUNT(DISTINCT user_id)"
454 } else {
455 "COUNT(*)"
456 };
457 let mut sql = format!(
458 "SELECT {col}, {cnt} FROM events \
459 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
460 AND {col} IS NOT NULL AND {col} != ''",
461 col = column,
462 cnt = count_expr,
463 );
464 if distinct_users {
465 sql.push_str(" AND user_id IS NOT NULL");
466 }
467 if event.is_some() {
468 sql.push_str(" AND event = ?");
469 }
470 let (extra_sql, extra_bind) = filter_clause(filter_url);
471 sql.push_str(extra_sql);
472 sql.push_str(&format!(" GROUP BY {col} ORDER BY {cnt} DESC LIMIT ?", col = column, cnt = count_expr));
473
474 let mut q = sqlx::query_as::<_, (String, i64)>(&sql)
475 .bind(property_id.as_bytes().to_vec())
476 .bind(start_ms)
477 .bind(end_ms);
478 if let Some(e) = event {
479 q = q.bind(e);
480 }
481 if let Some(v) = extra_bind {
482 q = q.bind(v);
483 }
484 q = q.bind(limit);
485 let rows = q.fetch_all(pool).await.unwrap_or_default();
486 rows.into_iter()
487 .map(|(label, count)| LabelCount { label, count })
488 .collect()
489}
490
491pub async fn events_by_screen_size(
492 pool: &SqlitePool,
493 property_id: &Uuid,
494 start_ms: i64,
495 end_ms: i64,
496 filter_url: Option<&str>,
497 limit: i64,
498) -> Vec<LabelCount> {
499 // Counts unique anonymous users (cookie-based user_id) per screen size,
500 // not raw events. Filtered to page_view so returning visitors are counted
501 // — the collectoruserid cookie suppresses session_start after the first
502 // visit, but page_view always fires.
503 let mut sql = String::from(
504 "SELECT screen_width, screen_height, COUNT(DISTINCT user_id) FROM events \
505 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
506 AND event = 'page_view' \
507 AND screen_width IS NOT NULL \
508 AND user_id IS NOT NULL",
509 );
510 let (extra_sql, extra_bind) = filter_clause(filter_url);
511 sql.push_str(extra_sql);
512 sql.push_str(" GROUP BY screen_width, screen_height ORDER BY COUNT(DISTINCT user_id) DESC LIMIT ?");
513 let mut q = sqlx::query_as::<_, (Option<i64>, Option<i64>, i64)>(&sql)
514 .bind(property_id.as_bytes().to_vec())
515 .bind(start_ms)
516 .bind(end_ms);
517 if let Some(v) = extra_bind {
518 q = q.bind(v);
519 }
520 q = q.bind(limit);
521 q.fetch_all(pool)
522 .await
523 .unwrap_or_default()
524 .into_iter()
525 .map(|(w, h, c)| LabelCount {
526 label: format!("{}x{}", w.unwrap_or(0), h.unwrap_or(0)),
527 count: c,
528 })
529 .collect()
530}
531
532// Device/browser/platform breakdowns filter on page_view (not session_start)
533// so they populate for returning visitors too. Server-side UA parsing fills
534// these columns on every event, so the data is always present.
535pub async fn events_by_device(pool: &SqlitePool, property_id: &Uuid, start_ms: i64, end_ms: i64, filter_url: Option<&str>, limit: i64) -> Vec<LabelCount> {
536 top_by_column(pool, property_id, start_ms, end_ms, filter_url, "device", Some("page_view"), limit, true).await
537}
538pub async fn events_by_browser(pool: &SqlitePool, property_id: &Uuid, start_ms: i64, end_ms: i64, filter_url: Option<&str>, limit: i64) -> Vec<LabelCount> {
539 top_by_column(pool, property_id, start_ms, end_ms, filter_url, "browser", Some("page_view"), limit, true).await
540}
541pub async fn events_by_platform(pool: &SqlitePool, property_id: &Uuid, start_ms: i64, end_ms: i64, filter_url: Option<&str>, limit: i64) -> Vec<LabelCount> {
542 top_by_column(pool, property_id, start_ms, end_ms, filter_url, "platform", Some("page_view"), limit, true).await
543}
544pub async fn events_by_page_url(pool: &SqlitePool, property_id: &Uuid, start_ms: i64, end_ms: i64, filter_url: Option<&str>, limit: i64) -> Vec<LabelCount> {
545 top_by_column(pool, property_id, start_ms, end_ms, filter_url, "url", None, limit, false).await
546}
547pub async fn page_views_by_page_url(pool: &SqlitePool, property_id: &Uuid, start_ms: i64, end_ms: i64, filter_url: Option<&str>, limit: i64) -> Vec<LabelCount> {
548 top_by_column(pool, property_id, start_ms, end_ms, filter_url, "url", Some("page_view"), limit, false).await
549}
550pub async fn session_starts_by_referrer(pool: &SqlitePool, property_id: &Uuid, start_ms: i64, end_ms: i64, filter_url: Option<&str>, limit: i64) -> Vec<LabelCount> {
551 top_by_column(pool, property_id, start_ms, end_ms, filter_url, "referrer", Some("session_start"), limit, false).await
552}
553
554pub async fn page_views_by_utm(
555 pool: &SqlitePool,
556 property_id: &Uuid,
557 start_ms: i64,
558 end_ms: i64,
559 filter_url: Option<&str>,
560 field: &str,
561 limit: i64,
562) -> Vec<LabelCount> {
563 let column = match field {
564 "source" => "utm_source",
565 "medium" => "utm_medium",
566 "campaign" => "utm_campaign",
567 "term" => "utm_term",
568 "content" => "utm_content",
569 _ => return Vec::new(),
570 };
571 top_by_column(pool, property_id, start_ms, end_ms, filter_url, column, Some("page_view"), limit, false).await
572}
573
574pub async fn events_by_custom_event(
575 pool: &SqlitePool,
576 property_id: &Uuid,
577 start_ms: i64,
578 end_ms: i64,
579 filter_url: Option<&str>,
580 limit: i64,
581) -> Vec<LabelCount> {
582 let placeholders = std::iter::repeat("?")
583 .take(BUILT_IN_EVENTS.len())
584 .collect::<Vec<_>>()
585 .join(",");
586 let (extra_sql, extra_bind) = filter_clause(filter_url);
587 let sql = format!(
588 "SELECT event, COUNT(*) FROM events \
589 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
590 AND event NOT IN ({}){} \
591 GROUP BY event ORDER BY COUNT(*) DESC LIMIT ?",
592 placeholders, extra_sql
593 );
594 let mut q = sqlx::query_as::<_, (String, i64)>(&sql)
595 .bind(property_id.as_bytes().to_vec())
596 .bind(start_ms)
597 .bind(end_ms);
598 for built in BUILT_IN_EVENTS {
599 q = q.bind(built);
600 }
601 if let Some(v) = extra_bind {
602 q = q.bind(v);
603 }
604 q = q.bind(limit);
605 q.fetch_all(pool)
606 .await
607 .unwrap_or_default()
608 .into_iter()
609 .map(|(label, count)| LabelCount { label, count })
610 .collect()
611}
612
613pub async fn session_starts_by_country(
614 pool: &SqlitePool,
615 property_id: &Uuid,
616 start_ms: i64,
617 end_ms: i64,
618 filter_url: Option<&str>,
619) -> std::collections::HashMap<String, i64> {
620 let (extra_sql, extra_bind) = filter_clause(filter_url);
621 let sql = format!(
622 "SELECT country, COUNT(*) FROM events \
623 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
624 AND event = 'session_start' AND country IS NOT NULL{} \
625 GROUP BY country",
626 extra_sql
627 );
628 let mut q = sqlx::query_as::<_, (String, i64)>(&sql)
629 .bind(property_id.as_bytes().to_vec())
630 .bind(start_ms)
631 .bind(end_ms);
632 if let Some(v) = extra_bind {
633 q = q.bind(v);
634 }
635 q.fetch_all(pool).await.unwrap_or_default().into_iter().collect()
636}
637
638pub async fn session_starts_by_country_region(
639 pool: &SqlitePool,
640 property_id: &Uuid,
641 start_ms: i64,
642 end_ms: i64,
643 filter_url: Option<&str>,
644) -> std::collections::HashMap<String, std::collections::HashMap<String, i64>> {
645 let (extra_sql, extra_bind) = filter_clause(filter_url);
646 let sql = format!(
647 "SELECT country, region, COUNT(*) FROM events \
648 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
649 AND event = 'session_start' \
650 AND country IS NOT NULL AND region IS NOT NULL{} \
651 GROUP BY country, region",
652 extra_sql
653 );
654 let mut q = sqlx::query_as::<_, (String, String, i64)>(&sql)
655 .bind(property_id.as_bytes().to_vec())
656 .bind(start_ms)
657 .bind(end_ms);
658 if let Some(v) = extra_bind {
659 q = q.bind(v);
660 }
661 let rows = q.fetch_all(pool).await.unwrap_or_default();
662 let mut out: std::collections::HashMap<String, std::collections::HashMap<String, i64>> =
663 std::collections::HashMap::new();
664 for (country, region, count) in rows {
665 out.entry(country).or_default().insert(region, count);
666 }
667 out
668}
669
670pub async fn bot_traffic(
671 pool: &SqlitePool,
672 property_id: &Uuid,
673 start_ms: i64,
674 end_ms: i64,
675 limit: i64,
676) -> BotTraffic {
677 let total: i64 = sqlx::query_scalar(
678 "SELECT COUNT(*) FROM bot_events \
679 WHERE property_id = ? AND created_at >= ? AND created_at <= ?",
680 )
681 .bind(property_id.as_bytes().to_vec())
682 .bind(start_ms)
683 .bind(end_ms)
684 .fetch_one(pool)
685 .await
686 .unwrap_or(0);
687 if total == 0 {
688 return BotTraffic::default();
689 }
690 let top_bots = sqlx::query_as::<_, (String, i64)>(
691 "SELECT bot_name, COUNT(*) FROM bot_events \
692 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
693 AND bot_name IS NOT NULL AND bot_name != '' \
694 GROUP BY bot_name ORDER BY COUNT(*) DESC LIMIT ?",
695 )
696 .bind(property_id.as_bytes().to_vec())
697 .bind(start_ms)
698 .bind(end_ms)
699 .bind(limit)
700 .fetch_all(pool)
701 .await
702 .unwrap_or_default()
703 .into_iter()
704 .map(|(label, count)| LabelCount { label, count })
705 .collect();
706 let top_pages = sqlx::query_as::<_, (String, i64)>(
707 "SELECT url, COUNT(*) FROM bot_events \
708 WHERE property_id = ? AND created_at >= ? AND created_at <= ? \
709 AND url IS NOT NULL AND url != '' \
710 GROUP BY url ORDER BY COUNT(*) DESC LIMIT ?",
711 )
712 .bind(property_id.as_bytes().to_vec())
713 .bind(start_ms)
714 .bind(end_ms)
715 .bind(limit)
716 .fetch_all(pool)
717 .await
718 .unwrap_or_default()
719 .into_iter()
720 .map(|(label, count)| LabelCount { label, count })
721 .collect();
722 BotTraffic { total, top_bots, top_pages }
723}
724
725/// Convert "YYYY-MM-DD" + a time-of-day to a unix-ms timestamp in the local tz.
726pub fn parse_date_to_ms(date: &str, end_of_day: bool) -> Option<i64> {
727 let nd = NaiveDate::parse_from_str(date, "%Y-%m-%d").ok()?;
728 let nt = if end_of_day {
729 chrono::NaiveTime::from_hms_opt(23, 59, 59)?
730 } else {
731 chrono::NaiveTime::from_hms_opt(0, 0, 0)?
732 };
733 let local: DateTime<chrono::Local> = chrono::Local
734 .from_local_datetime(&chrono::NaiveDateTime::new(nd, nt))
735 .single()?;
736 Some(local.with_timezone(&Utc).timestamp_millis())
737}