use std::collections::hash_map::RandomState;
use std::collections::{BTreeMap, HashMap, VecDeque};
use std::hash::{BuildHasher, Hash, Hasher};
use std::net::IpAddr;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock, RwLock};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use super::{classify, Class, Group, Kind};
pub const MAX_HOSTS: usize = 16;
pub const TOP: usize = 200;
pub const TOP_SAVED: usize = 100;
pub const MAX_PENDING: usize = 4096;
pub const MAX_CACHE: usize = 16384;
pub const UNDECIDED: Duration = Duration::from_secs(120);
const MAX_PENDING_REQUESTS: usize = 32;
const SHARDS: usize = 16;
const PEOPLE: usize = 0;
const AI: usize = 1;
const BOTS: usize = 2;
fn slot(group: Group) -> Option<usize> {
match group {
Group::Person => Some(PEOPLE),
Group::Ai => Some(AI),
Group::Bot => Some(BOTS),
Group::Internal => None,
}
}
#[derive(Clone, Debug)]
#[doc(hidden)]
pub struct Hit<'a> {
pub ip: IpAddr,
pub host: &'a str,
pub user_agent: Option<&'a str>,
pub referer: Option<&'a str>,
pub method: &'a str,
pub path: &'a str,
pub status: i16,
pub content_type: Option<&'a str>,
pub headers: &'a [crate::header::Header],
pub now: u64,
}
pub const DIMS: [&str; 8] = ["countries", "browsers", "os", "devices", "channels", "utm_source", "utm_medium", "utm_campaign"];
const COUNTRIES: usize = 0;
const CHANNELS: usize = 4;
const UTM: usize = 5;
pub const SESSION_GAP: u64 = 30 * 60;
#[cfg(not(test))]
pub const MAX_OPEN_SESSIONS: usize = 8192;
#[cfg(test)]
pub const MAX_OPEN_SESSIONS: usize = 64;
pub const BEHAVIOUR: [&str; 7] = ["engaged_secs", "engaged_views", "scroll", "screens", "events", "event_props", "goals"];
const ENGAGED_SECS: usize = 0;
const ENGAGED_VIEWS: usize = 1;
const SCROLL: usize = 2;
const SCREENS: usize = 3;
const EVENTS: usize = 4;
const EVENT_PROPS: usize = 5;
const GOALS: usize = 6;
pub const MAX_REPORTS: u32 = 500;
pub const MAX_ENGAGED_SECS: u64 = 6 * 3600;
pub const LIVE_WINDOW: u64 = 30 * 60;
pub const MAX_RECENT: usize = 4096;
#[derive(Clone, Debug)]
#[doc(hidden)]
pub struct Recent {
pub at: u64,
pub visitor: u64,
pub page: String,
pub country: Option<[u8; 2]>,
}
pub const DIM_TOP: usize = 400;
pub const DIM_SAVED: usize = 300;
#[derive(Clone, Debug)]
struct Audience {
country: Option<[u8; 2]>,
device: super::device::Device,
}
#[derive(Clone, Debug, Default)]
struct Arrival {
channel: &'static str,
utm: [Option<String>; 3],
}
#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)]
#[doc(hidden)]
pub struct Counts {
pub requests: u64,
pub pages: u64,
}
#[derive(Default, Debug, Clone)]
#[doc(hidden)]
pub struct Top(pub HashMap<String, u64>);
impl Top {
fn add(&mut self, key: &str, n: u64) {
self.add_capped(key, n, TOP);
}
fn add_capped(&mut self, key: &str, n: u64, cap: usize) {
if let Some(count) = self.0.get_mut(key) {
*count += n;
} else if self.0.len() < cap {
self.0.insert(key.chars().take(300).collect(), n);
} else {
*self.0.entry("(other)".to_string()).or_default() += n;
}
}
#[doc(hidden)]
pub fn merge(&mut self, other: &Top) {
for (key, n) in &other.0 {
*self.0.entry(key.clone()).or_default() += n;
}
}
#[doc(hidden)]
pub fn sorted(&self, n: usize) -> Vec<(String, u64)> {
let mut out: Vec<(String, u64)> = self.0.iter().map(|(k, v)| (k.clone(), *v)).collect();
out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
out.truncate(n);
out
}
}
#[derive(Default, Debug, Clone)]
#[doc(hidden)]
pub struct Day {
pub visitors: [u64; 3],
pub people_pages: Top,
pub ai_pages: Top,
pub ai_agents: Top,
pub bots: Top,
pub referrers: Top,
pub not_found: Top,
pub dims: [Top; 8],
pub sessions: BTreeMap<String, Sessions>,
pub entry_pages: Top,
pub exit_pages: Top,
pub behaviour: [Top; 7],
}
#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)]
#[doc(hidden)]
pub struct Sessions {
pub sessions: u64,
pub bounces: u64,
pub duration: u64,
pub pages: u64,
}
impl Sessions {
#[doc(hidden)]
pub fn add(&mut self, other: &Sessions) {
self.sessions += other.sessions;
self.bounces += other.bounces;
self.duration += other.duration;
self.pages += other.pages;
}
}
struct Session {
start: u64,
last: u64,
pages: u64,
exit: String,
goals: u32,
channel: Option<&'static str>,
utm: [Option<String>; 3],
reports: u32,
}
impl Session {
fn totals(&self) -> Sessions {
Sessions { sessions: 1, bounces: (self.pages == 1) as u64, duration: self.last - self.start, pages: self.pages }
}
}
impl Day {
#[doc(hidden)]
pub fn merge(&mut self, other: &Day) {
for g in 0..3 {
self.visitors[g] += other.visitors[g];
}
for (mine, theirs) in self.dims.iter_mut().zip(&other.dims) {
mine.merge(theirs);
}
self.people_pages.merge(&other.people_pages);
self.ai_pages.merge(&other.ai_pages);
self.ai_agents.merge(&other.ai_agents);
self.bots.merge(&other.bots);
self.referrers.merge(&other.referrers);
self.not_found.merge(&other.not_found);
for (mine, theirs) in self.behaviour.iter_mut().zip(&other.behaviour) {
mine.merge(theirs);
}
self.entry_pages.merge(&other.entry_pages);
self.exit_pages.merge(&other.exit_pages);
for (host, s) in &other.sessions {
self.sessions.entry(host.clone()).or_default().add(s);
}
}
}
struct Pending {
last: u64,
requests: Vec<Deferred>,
}
struct Deferred {
at: u64,
hour: u64,
host: u8,
page: Option<String>,
referrer: Option<String>,
not_found: Option<String>,
arrival: Option<Arrival>,
}
#[derive(Clone)]
struct Cached {
class: Class,
audience: Option<Audience>,
}
#[derive(Default)]
struct Shard {
hours: BTreeMap<(u64, u8), [Counts; 3]>,
days: BTreeMap<u64, Day>,
seen: HashMap<u64, Group>,
pending: HashMap<u64, Pending>,
cache: HashMap<u64, Cached>,
open: HashMap<u64, Session>,
recent: VecDeque<Recent>,
today: u64,
}
struct Recorder {
shards: Vec<Mutex<Shard>>,
hosts: RwLock<Vec<String>>,
secret: RandomState,
retention_days: u64,
exclude: Vec<String>,
}
static ENABLED: AtomicBool = AtomicBool::new(false);
static HOSTS_GENERATION: AtomicU64 = AtomicU64::new(1);
static CALLS: AtomicU64 = AtomicU64::new(0);
fn recorder() -> &'static Recorder {
static RECORDER: OnceLock<Recorder> = OnceLock::new();
RECORDER.get_or_init(|| Recorder {
shards: (0..SHARDS).map(|_| Mutex::new(Shard::default())).collect(),
hosts: RwLock::new(Vec::new()),
secret: RandomState::new(),
retention_days: super::config::retention_days(),
exclude: super::config::exclude(),
})
}
pub fn enable() {
#[cfg(not(target_arch = "wasm32"))]
crate::log::set_access_observer(super::observe);
let _ = recorder();
ENABLED.store(true, Ordering::Relaxed);
}
pub fn enabled() -> bool {
ENABLED.load(Ordering::Relaxed)
}
pub fn now() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
}
fn normalize_host(host: &str) -> String {
let host = host.trim();
let host = match host.strip_prefix('[') {
Some(rest) => rest.split(']').next().unwrap_or(rest),
None => host.split(':').next().unwrap_or(host),
};
if host.is_empty() { "(none)".to_string() } else { host.to_ascii_lowercase() }
}
fn host_index(raw: &str) -> u8 {
thread_local! {
static LAST: std::cell::RefCell<(u64, String, u8)> = const { std::cell::RefCell::new((0, String::new(), 0)) };
}
let generation = HOSTS_GENERATION.load(Ordering::Relaxed);
if let Ok(Some(i)) = LAST.try_with(|last| {
let last = last.borrow();
(last.0 == generation && !last.1.is_empty() && last.1 == raw).then_some(last.2)
}) {
return i;
}
let i = host_index_slow(raw);
let _ = LAST.try_with(|last| *last.borrow_mut() = (generation, raw.to_string(), i));
i
}
fn host_index_slow(host: &str) -> u8 {
let host = normalize_host(host);
let recorder = recorder();
if let Some(i) = recorder.hosts.read().unwrap_or_else(|e| e.into_inner()).iter().position(|h| *h == host) {
return i as u8;
}
let mut hosts = recorder.hosts.write().unwrap_or_else(|e| e.into_inner());
if let Some(i) = hosts.iter().position(|h| *h == host) {
return i as u8;
}
if hosts.len() < MAX_HOSTS {
hosts.push(host);
return (hosts.len() - 1) as u8;
}
if let Some(i) = hosts.iter().position(|h| h == "other") {
return i as u8;
}
hosts.push("other".to_string());
(hosts.len() - 1) as u8
}
#[doc(hidden)]
pub fn host_names() -> Vec<String> {
recorder().hosts.read().unwrap_or_else(|e| e.into_inner()).clone()
}
fn visitor(day: u64, host: u8, ip: IpAddr, user_agent: Option<&str>) -> u64 {
let mut hasher = recorder().secret.build_hasher();
day.hash(&mut hasher);
host.hash(&mut hasher);
ip.hash(&mut hasher);
user_agent.unwrap_or("").hash(&mut hasher);
hasher.finish()
}
fn is_page(hit: &Hit<'_>) -> bool {
if hit.method != "GET" || !matches!(hit.status, 200 | 304) {
return false;
}
match hit.content_type {
Some(ct) => ct.trim_start().as_bytes().get(..9).is_some_and(|p| p.eq_ignore_ascii_case(b"text/html")),
None => {
let last = hit.path.rsplit('/').next().unwrap_or("");
last.is_empty() || !last.contains('.') || last.ends_with(".html")
}
}
}
fn is_asset(path: &str) -> bool {
let path = path.split('?').next().unwrap_or(path).as_bytes();
[&b".css"[..], b".js", b".mjs", b".woff2", b".woff"].iter().any(|ext| path.len() >= ext.len() && path[path.len() - ext.len()..].eq_ignore_ascii_case(ext))
}
fn page_key(host: &str, path: &str) -> String {
format!("{}{}", normalize_host(host), path.split('?').next().unwrap_or(path))
}
fn referrer_host(referer: Option<&str>, host: &str) -> Option<String> {
let rest = referer?.split_once("://")?.1;
let other = rest.split(['/', '?', '#']).next()?.split(':').next()?.to_ascii_lowercase();
(!other.is_empty() && other != normalize_host(host)).then_some(other)
}
#[doc(hidden)]
pub fn channel_of(referrer_host: &str) -> &'static str {
let h = referrer_host.strip_prefix("www.").unwrap_or(referrer_host);
let is = |d: &str| h == d || h.ends_with(&format!(".{d}"));
const AI: &[&str] = &["chatgpt.com", "chat.openai.com", "perplexity.ai", "claude.ai", "gemini.google.com", "copilot.microsoft.com", "chat.deepseek.com", "chat.mistral.ai", "you.com", "phind.com", "kagi.com"];
const SEARCH: &[&str] = &["bing.com", "duckduckgo.com", "search.yahoo.com", "yandex.ru", "yandex.com", "baidu.com", "ecosia.org", "search.brave.com", "startpage.com", "qwant.com", "naver.com", "seznam.cz"];
const SOCIAL: &[&str] = &[
"facebook.com", "m.facebook.com", "l.facebook.com", "lm.facebook.com", "t.co", "twitter.com", "x.com", "linkedin.com", "lnkd.in", "reddit.com", "instagram.com",
"youtube.com", "tiktok.com", "pinterest.com", "news.ycombinator.com", "mastodon.social", "bsky.app", "threads.net", "vk.com", "telegram.org", "t.me", "discord.com",
"whatsapp.com", "lobste.rs",
];
if AI.iter().any(|d| is(d)) {
"ai"
} else if SEARCH.iter().any(|d| is(d)) || h.starts_with("google.") || h.contains(".google.") {
"search"
} else if SOCIAL.iter().any(|d| is(d)) {
"social"
} else {
"referral"
}
}
fn query_value(raw: &str) -> Option<String> {
let bytes = raw.as_bytes();
let mut out = Vec::with_capacity(bytes.len().min(300));
let mut i = 0;
while i < bytes.len() && out.len() < 300 {
match bytes[i] {
b'+' => out.push(b' '),
b'%' if i + 2 < bytes.len() => {
match std::str::from_utf8(&bytes[i + 1..i + 3]).ok().and_then(|h| u8::from_str_radix(h, 16).ok()) {
Some(b) => {
out.push(b);
i += 2;
}
None => out.push(b'%'),
}
}
b => out.push(b),
}
i += 1;
}
let value: String = String::from_utf8_lossy(&out).trim().chars().filter(|c| !c.is_control()).take(100).collect();
(!value.is_empty()).then_some(value)
}
#[doc(hidden)]
pub fn utm(target: &str) -> [Option<String>; 3] {
let mut out: [Option<String>; 3] = Default::default();
let Some(query) = target.split_once('?').map(|(_, q)| q.split('#').next().unwrap_or(q)) else { return out };
for pair in query.split('&') {
let (name, value) = pair.split_once('=').unwrap_or((pair, ""));
let slot = match name {
"utm_source" => 0,
"utm_medium" => 1,
"utm_campaign" => 2,
_ => continue,
};
if out[slot].is_none() {
out[slot] = query_value(value);
}
}
out
}
fn arrival(hit: &Hit<'_>) -> Option<Arrival> {
let utm = utm(hit.path);
let channel = if utm[0].is_some() || utm[1].is_some() || utm[2].is_some() {
"campaign"
} else {
match hit.referer {
None => "direct",
Some(_) => channel_of(&referrer_host(hit.referer, hit.host)?),
}
};
Some(Arrival { channel, utm })
}
#[derive(Default)]
struct Local {
decided: HashMap<u64, (u64, usize)>,
counts: BTreeMap<(u64, u8), [Counts; 3]>,
}
const LOCAL_DECIDED: usize = 1024;
fn locals() -> &'static Mutex<Vec<std::sync::Weak<Mutex<Local>>>> {
static LOCALS: OnceLock<Mutex<Vec<std::sync::Weak<Mutex<Local>>>>> = OnceLock::new();
LOCALS.get_or_init(Mutex::default)
}
thread_local! {
static LOCAL: std::sync::Arc<Mutex<Local>> = {
let local = std::sync::Arc::new(Mutex::new(Local::default()));
let mut all = locals().lock().unwrap_or_else(|e| e.into_inner());
all.retain(|w| w.strong_count() > 0);
all.push(std::sync::Arc::downgrade(&local));
local
};
}
fn drain_locals() {
let all: Vec<_> = locals().lock().unwrap_or_else(|e| e.into_inner()).iter().filter_map(|w| w.upgrade()).collect();
let mut target = recorder().shards[0].lock().unwrap_or_else(|e| e.into_inner());
for local in all {
let counts = std::mem::take(&mut local.lock().unwrap_or_else(|e| e.into_inner()).counts);
for (key, c) in counts {
let entry = target.hours.entry(key).or_default();
for g in 0..3 {
entry[g].requests += c[g].requests;
entry[g].pages += c[g].pages;
}
}
}
}
#[doc(hidden)]
pub fn record(hit: &Hit<'_>) {
if !enabled() {
return;
}
let recorder = recorder();
let path = hit.path.split('?').next().unwrap_or(hit.path);
if recorder.exclude.iter().any(|e| path == e) || path == super::script::BEACON_PATH {
return;
}
let day = hit.now / 86_400;
let hour = hit.now / 3_600;
let host = host_index(hit.host);
let visitor = visitor(day, host, hit.ip, hit.user_agent);
if hit.status != 404 && !is_page(hit) {
let counted = LOCAL.try_with(|local| {
let mut local = local.lock().unwrap_or_else(|e| e.into_inner());
match local.decided.get(&visitor) {
Some((d, g)) if *d == day => {
let g = *g;
local.counts.entry((hour, host)).or_default()[g].requests += 1;
true
}
_ => false,
}
});
if counted == Ok(true) {
return;
}
}
let decided = record_shared(hit, day, hour, host, visitor, path);
if let Some(g) = decided {
let _ = LOCAL.try_with(|local| {
let mut local = local.lock().unwrap_or_else(|e| e.into_inner());
if local.decided.len() >= LOCAL_DECIDED {
local.decided.clear();
}
local.decided.insert(visitor, (day, g));
});
}
if CALLS.fetch_add(1, Ordering::Relaxed) % 1024 == 0 {
settle(hit.now);
}
}
fn record_shared(hit: &Hit<'_>, day: u64, hour: u64, host: u8, visitor: u64, path: &str) -> Option<usize> {
let recorder = recorder();
let mut shard = recorder.shards[(visitor % SHARDS as u64) as usize].lock().unwrap_or_else(|e| e.into_inner());
shard.roll(day, recorder.retention_days);
let cached = match shard.cache.get(&visitor) {
Some(cached) => cached.clone(),
None => {
let class = classify(hit.ip, hit.user_agent, "/");
let audience = (class.kind == Kind::Browser).then(|| Audience {
country: super::geo::country(hit.ip),
device: {
let hint = |name: &str| hit.headers.iter().find(|h| h.name.eq_ignore_ascii_case(name)).map(|h| h.value.as_ref());
super::device::parse(hit.user_agent.unwrap_or(""), hint("sec-ch-ua-platform"), hint("sec-ch-ua-mobile"))
},
});
let cached = Cached { class, audience };
if shard.cache.len() >= MAX_CACHE {
shard.cache.clear();
}
shard.cache.insert(visitor, cached.clone());
cached
}
};
let class = cached.class;
let scanner = super::classify::scanner_path(hit.path) && matches!(class.kind, Kind::Browser | Kind::Tool | Kind::NoUserAgent);
let class = if scanner { Class { kind: Kind::Scanner, name: "", operator: "", verified: false } } else { class };
let page = is_page(hit).then(|| page_key(hit.host, path));
let not_found = (hit.status == 404).then(|| page_key(hit.host, path));
let referrer = referrer_host(hit.referer, hit.host);
let arrival = if page.is_some() && class.kind == Kind::Browser { arrival(hit) } else { None };
let decided = match class.group() {
Group::Internal => None,
Group::Person => {
match shard.seen.get(&visitor).copied() {
Some(Group::Person) => {
shard.count_person(visitor, hit.now, host, page, referrer, not_found, arrival);
Some(PEOPLE)
}
Some(_) => {
shard.count(BOTS, hour, host, &browser_without_assets(), page, not_found, visitor);
Some(BOTS)
}
None if is_asset(path) => {
shard.seen.insert(visitor, Group::Person);
shard.day(day).visitors[PEOPLE] += 1;
if let Some(audience) = &cached.audience {
shard.count_audience(day, &normalize_host(hit.host), audience);
}
if let Some(pending) = shard.pending.remove(&visitor) {
for d in pending.requests {
shard.count_person(visitor, d.at, d.host, d.page, d.referrer, d.not_found, d.arrival);
}
}
shard.count_person(visitor, hit.now, host, page, referrer, not_found, arrival);
Some(PEOPLE)
}
None => {
if shard.pending.len() >= MAX_PENDING && !shard.pending.contains_key(&visitor) {
shard.seen.insert(visitor, Group::Bot);
shard.day(day).visitors[BOTS] += 1;
shard.count(BOTS, hour, host, &browser_without_assets(), page, not_found, visitor);
Some(BOTS)
} else {
let pending = shard.pending.entry(visitor).or_insert(Pending { last: hit.now, requests: Vec::new() });
pending.last = hit.now;
pending.requests.push(Deferred { at: hit.now, hour, host, page, referrer, not_found, arrival });
if pending.requests.len() < MAX_PENDING_REQUESTS {
None
} else {
let pending = shard.pending.remove(&visitor).expect("just added");
shard.seen.insert(visitor, Group::Bot);
shard.day(day).visitors[BOTS] += 1;
for d in pending.requests {
shard.count(BOTS, d.hour, d.host, &browser_without_assets(), d.page, d.not_found, visitor);
}
Some(BOTS)
}
}
}
}
}
group => {
let g = slot(group).expect("counted group");
if shard.seen.insert(visitor, group).is_none() {
shard.day(day).visitors[g] += 1;
}
shard.count(g, hour, host, &class, page, not_found, visitor);
(!scanner).then_some(g)
}
};
decided
}
fn browser_without_assets() -> Class {
Class { kind: Kind::Browser, name: "browser, no assets", operator: "", verified: false }
}
impl Shard {
fn day(&mut self, day: u64) -> &mut Day {
self.days.entry(day).or_default()
}
fn roll(&mut self, day: u64, retention_days: u64) {
if self.today == day {
return;
}
self.today = day;
self.close_sessions(|_| true);
self.seen.clear();
self.cache.clear();
let oldest_day = day.saturating_sub(retention_days.saturating_sub(1));
self.days.retain(|d, _| *d >= oldest_day);
self.hours.retain(|(h, _), _| *h / 24 >= oldest_day);
}
#[cfg(feature = "model-postgres")]
fn store_view(&self, visitor: u64, at: u64, page: &str, referrer: Option<&str>) {
let (host, path) = page.find('/').map_or((page, "/"), |i| (&page[..i], &page[i..]));
let session = self.open.get(&visitor);
let audience = self.cache.get(&visitor).and_then(|c| c.audience.as_ref());
super::store::enqueue(super::store::Row::PageView(super::store::PageView {
at,
host: host.to_string(),
path: path.to_string(),
visitor,
session_start: session.map_or(at, |s| s.start),
referrer_host: referrer.map(str::to_string),
channel: session.and_then(|s| s.channel),
utm: session.map(|s| s.utm.clone()).unwrap_or_default(),
country: audience.and_then(|a| a.country).map(|c| String::from_utf8_lossy(&c).into_owned()),
browser: audience.map(|a| a.device.browser),
os: audience.map(|a| a.device.os),
device: audience.map(|a| a.device.class),
}));
}
fn count_audience(&mut self, day: u64, host: &str, audience: &Audience) {
let day = self.day(day);
let country = audience.country.map(|c| String::from_utf8_lossy(&c).into_owned());
let values = [country.as_deref().unwrap_or("(unknown)"), audience.device.browser, audience.device.os, audience.device.class];
for (i, value) in values.iter().enumerate() {
day.dims[COUNTRIES + i].add_capped(&format!("{host}\t{value}"), 1, DIM_TOP);
}
}
fn close_sessions(&mut self, which: impl Fn(&Session) -> bool) {
let done: Vec<u64> = self.open.iter().filter(|(_, s)| which(s)).map(|(v, _)| *v).collect();
for visitor in done {
if let Some(session) = self.open.remove(&visitor) {
self.close(session);
}
}
}
fn close(&mut self, session: Session) {
let day = self.day(session.start / 86_400);
add_session(day, &session);
}
fn session_page(&mut self, visitor: u64, at: u64, page: &str) -> bool {
if let Some(session) = self.open.get_mut(&visitor) {
if at.saturating_sub(session.last) < SESSION_GAP {
session.last = session.last.max(at);
session.pages += 1;
session.exit = page.to_string();
session.goals |= page_goals(page);
return false;
}
let ended = self.open.remove(&visitor).expect("just found");
self.close(ended);
}
if self.open.len() >= MAX_OPEN_SESSIONS {
if let Some(quietest) = self.open.iter().min_by_key(|(_, s)| s.last).map(|(v, _)| *v) {
let ended = self.open.remove(&quietest).expect("just found");
self.close(ended);
}
}
self.day(at / 86_400).entry_pages.add(page, 1);
self.open.insert(visitor, Session { start: at, last: at, pages: 1, exit: page.to_string(), goals: 0, reports: 0, channel: None, utm: Default::default() });
if let Some(session) = self.open.get_mut(&visitor) {
session.goals |= page_goals(page);
}
true
}
#[allow(clippy::too_many_arguments)]
fn count_person(&mut self, visitor: u64, at: u64, host: u8, page: Option<String>, referrer: Option<String>, not_found: Option<String>, arrival: Option<Arrival>) {
let hour = at / 3_600;
let started = page.as_ref().is_some_and(|page| self.session_page(visitor, at, page));
if let Some(page) = &page {
let country = self.cache.get(&visitor).and_then(|c| c.audience.as_ref()).and_then(|a| a.country);
while self.recent.front().is_some_and(|r| r.at + LIVE_WINDOW < at) || self.recent.len() >= MAX_RECENT {
self.recent.pop_front();
}
self.recent.push_back(Recent { at, visitor, page: page.clone(), country });
}
let arrival = arrival.filter(|_| started);
if let (Some(arrival), Some(session)) = (&arrival, self.open.get_mut(&visitor)) {
session.channel = Some(arrival.channel);
session.utm = arrival.utm.clone();
}
#[cfg(feature = "model-postgres")]
if let (true, Some(page)) = (super::store::enabled(), &page) {
self.store_view(visitor, at, page, referrer.as_deref());
}
let counts = &mut self.hours.entry((hour, host)).or_default()[PEOPLE];
counts.requests += 1;
counts.pages += page.is_some() as u64;
let day = self.day(hour / 24);
if let (Some(arrival), Some(page)) = (arrival, &page) {
let host = page.split('/').next().unwrap_or("");
day.dims[CHANNELS].add_capped(&format!("{host}\t{}", arrival.channel), 1, DIM_TOP);
for (i, value) in arrival.utm.iter().enumerate() {
if let Some(value) = value {
day.dims[UTM + i].add_capped(&format!("{host}\t{value}"), 1, DIM_TOP);
}
}
}
if let Some(page) = page {
day.people_pages.add(&page, 1);
if let Some(referrer) = referrer {
day.referrers.add(&referrer, 1);
}
}
if let Some(path) = not_found {
day.not_found.add(&path, 1);
}
}
#[allow(clippy::too_many_arguments)]
fn count(&mut self, g: usize, hour: u64, host: u8, class: &Class, page: Option<String>, not_found: Option<String>, _visitor: u64) {
let counts = &mut self.hours.entry((hour, host)).or_default()[g];
counts.requests += 1;
counts.pages += page.is_some() as u64;
let day = self.day(hour / 24);
if g == AI {
day.ai_agents.add(&format!("{}\t{}\t{}\t{}", or_unknown(class.name), class.operator, class.kind.label(), class.verified), 1);
if let Some(page) = page {
day.ai_pages.add(&page, 1);
}
} else {
let kind = if class.kind == Kind::Browser { "browser-no-assets" } else { class.kind.label() };
day.bots.add(&format!("{kind}\t{}", class.name), 1);
}
if let Some(path) = not_found {
day.not_found.add(&path, 1);
}
}
}
fn or_unknown(name: &str) -> &str {
if name.is_empty() { "(unnamed)" } else { name }
}
#[doc(hidden)]
pub fn settle(now: u64) {
for shard in &recorder().shards {
let mut shard = shard.lock().unwrap_or_else(|e| e.into_inner());
shard.close_sessions(|s| now.saturating_sub(s.last) >= SESSION_GAP);
let done: Vec<u64> = shard.pending.iter().filter(|(_, p)| now.saturating_sub(p.last) >= UNDECIDED.as_secs()).map(|(v, _)| *v).collect();
for visitor in done {
let Some(pending) = shard.pending.remove(&visitor) else { continue };
let day = pending.requests.first().map_or(now / 86_400, |d| d.hour / 24);
if shard.seen.insert(visitor, Group::Bot).is_none() {
shard.day(day).visitors[BOTS] += 1;
}
for d in pending.requests {
shard.count(BOTS, d.hour, d.host, &browser_without_assets(), d.page, d.not_found, visitor);
}
}
}
}
#[doc(hidden)]
pub fn snapshot(now: u64) -> (BTreeMap<(u64, u8), [Counts; 3]>, BTreeMap<u64, Day>) {
settle(now);
drain_locals();
let mut hours: BTreeMap<(u64, u8), [Counts; 3]> = BTreeMap::new();
let mut days: BTreeMap<u64, Day> = BTreeMap::new();
for shard in &recorder().shards {
let shard = shard.lock().unwrap_or_else(|e| e.into_inner());
for (key, counts) in &shard.hours {
let entry = hours.entry(*key).or_default();
for g in 0..3 {
entry[g].requests += counts[g].requests;
entry[g].pages += counts[g].pages;
}
}
for (day, lists) in &shard.days {
days.entry(*day).or_default().merge(lists);
}
for session in shard.open.values() {
add_session(days.entry(session.start / 86_400).or_default(), session);
}
}
(hours, days)
}
fn add_session(day: &mut Day, session: &Session) {
let host = session.exit.split('/').next().unwrap_or("").to_string();
day.sessions.entry(host.clone()).or_default().add(&session.totals());
day.exit_pages.add(&session.exit, 1);
if session.goals != 0 {
for (i, goal) in goals().iter().enumerate() {
if session.goals & (1 << i) != 0 {
day.behaviour[GOALS].add_capped(&format!("{host}\t{goal}"), 1, DIM_TOP);
}
}
}
}
fn goals_lock() -> &'static RwLock<Vec<String>> {
static GOALS_SET: OnceLock<RwLock<Vec<String>>> = OnceLock::new();
GOALS_SET.get_or_init(|| RwLock::new(super::config::goals()))
}
fn goals() -> std::sync::RwLockReadGuard<'static, Vec<String>> {
goals_lock().read().unwrap_or_else(|e| e.into_inner())
}
#[cfg(test)]
#[doc(hidden)]
pub fn set_goals_for_tests(goals: &[&str]) {
*goals_lock().write().unwrap_or_else(|e| e.into_inner()) = goals.iter().map(|g| g.to_string()).collect();
}
fn page_goals(page: &str) -> u32 {
let path = page.find('/').map_or("/", |i| &page[i..]);
let mut reached = 0;
for (i, goal) in goals().iter().enumerate() {
let hit = match goal.strip_suffix('*') {
_ if goal.starts_with("event:") => false,
Some(prefix) => path.starts_with(prefix),
None => path == goal,
};
if hit {
reached |= 1 << i;
}
}
reached
}
fn event_goals(name: &str) -> u32 {
let mut reached = 0;
for (i, goal) in goals().iter().enumerate() {
if goal.strip_prefix("event:") == Some(name) {
reached |= 1 << i;
}
}
reached
}
fn json_string(raw: &str) -> Option<String> {
crate::mcp::json_rpc::extract_str(&format!("{{\"v\":{raw}}}"), "v")
}
fn width_bucket(width: u64) -> &'static str {
match width {
0 => "(unknown)",
1..=599 => "phone",
600..=1023 => "tablet",
1024..=1439 => "laptop",
_ => "wide",
}
}
#[doc(hidden)]
pub fn beacon(ip: IpAddr, raw_host: &str, user_agent: Option<&str>, body: &str, now: u64) {
if !enabled() {
return;
}
let Some(members) = crate::mcp::json_rpc::object_members(body) else { return };
let get = |key: &str| members.iter().find(|(k, _)| k == key).map(|(_, v)| v.as_str());
let number = |key: &str| get(key).and_then(|v| v.trim().parse::<f64>().ok()).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u64);
let Some(kind) = get("t").and_then(json_string) else { return };
let Some(path) = get("p").and_then(json_string).filter(|p| p.starts_with('/') && p.len() <= 1000) else { return };
let host = host_index(raw_host);
let day = now / 86_400;
let visitor = visitor(day, host, ip, user_agent);
let host_name = normalize_host(raw_host);
if kind == "view" {
let from = get("r").and_then(json_string).filter(|r| r.starts_with('/')).unwrap_or_else(|| "/".to_string());
let referer = format!("https://{host_name}{from}");
let is_person = {
let shard = recorder().shards[(visitor % SHARDS as u64) as usize].lock().unwrap_or_else(|e| e.into_inner());
shard.seen.get(&visitor) == Some(&Group::Person)
};
if is_person {
record(&Hit {
ip,
host: raw_host,
user_agent,
referer: Some(&referer),
method: "GET",
path: &path,
status: 200,
content_type: Some("text/html"),
headers: &[],
now,
});
}
return;
}
let mut shard = recorder().shards[(visitor % SHARDS as u64) as usize].lock().unwrap_or_else(|e| e.into_inner());
if shard.seen.get(&visitor) != Some(&Group::Person) {
return;
}
if let Some(session) = shard.open.get_mut(&visitor) {
session.reports += 1;
if session.reports > MAX_REPORTS {
return;
}
}
let page = page_key(raw_host, &path);
match kind.as_str() {
"leave" => {
let Some(ms) = number("ms") else { return };
let secs = ms.min(MAX_ENGAGED_SECS * 1000) / 1000;
let depth = number("sd").unwrap_or(0).min(100);
let width = number("w").unwrap_or(0);
let bucket = match depth {
0..=25 => "25",
26..=50 => "50",
51..=75 => "75",
_ => "100",
};
#[cfg(feature = "model-postgres")]
super::store::enqueue(super::store::Row::Report(super::store::Report {
at: now,
host: host_name.clone(),
path: path.clone(),
visitor,
kind: "leave",
secs: Some(secs),
scroll: Some(depth),
width: Some(width),
..Default::default()
}));
let d = shard.day(day);
d.behaviour[ENGAGED_SECS].add(&page, secs);
d.behaviour[ENGAGED_VIEWS].add(&page, 1);
d.behaviour[SCROLL].add_capped(&format!("{host_name}\t{bucket}"), 1, DIM_TOP);
d.behaviour[SCREENS].add_capped(&format!("{host_name}\t{}", width_bucket(width)), 1, DIM_TOP);
}
"event" => {
let Some(name) = get("n").and_then(json_string).map(|n| n.chars().filter(|c| !c.is_control()).take(64).collect::<String>()).filter(|n| !n.is_empty()) else {
return;
};
let props: Vec<(String, String)> = get("q")
.and_then(crate::mcp::json_rpc::object_members)
.unwrap_or_default()
.into_iter()
.take(4)
.filter_map(|(k, v)| Some((k.chars().take(32).collect::<String>(), json_string(&v)?.chars().take(64).collect::<String>())))
.collect();
let reached = event_goals(&name);
if reached != 0 {
if let Some(session) = shard.open.get_mut(&visitor) {
session.goals |= reached;
}
}
#[cfg(feature = "model-postgres")]
super::store::enqueue(super::store::Row::Report(super::store::Report {
at: now,
host: host_name.clone(),
path: path.clone(),
visitor,
kind: "event",
name: Some(name.clone()),
props: (!props.is_empty()).then(|| props.iter().map(|(k, v)| format!("{k}={v}")).collect::<Vec<_>>().join("&")),
..Default::default()
}));
let d = shard.day(day);
d.behaviour[EVENTS].add_capped(&format!("{host_name}\t{name}"), 1, DIM_TOP);
for (k, v) in props {
d.behaviour[EVENT_PROPS].add_capped(&format!("{host_name}\t{name}\t{k}={v}"), 1, DIM_TOP);
}
}
_ => {}
}
}
#[doc(hidden)]
pub fn recent(now: u64) -> Vec<Recent> {
settle(now);
let mut out = Vec::new();
for shard in &recorder().shards {
let shard = shard.lock().unwrap_or_else(|e| e.into_inner());
out.extend(shard.recent.iter().filter(|r| r.at + LIVE_WINDOW >= now && r.at <= now).cloned());
}
out
}
#[doc(hidden)]
pub fn restore(hosts: Vec<String>, hours: BTreeMap<(u64, u8), [Counts; 3]>, days: BTreeMap<u64, Day>) {
let recorder = recorder();
let mut known = recorder.hosts.write().unwrap_or_else(|e| e.into_inner());
let map: Vec<u8> = hosts
.iter()
.map(|h| match known.iter().position(|k| k == h) {
Some(i) => i as u8,
None if known.len() < MAX_HOSTS + 1 => {
known.push(h.clone());
(known.len() - 1) as u8
}
None => 0,
})
.collect();
drop(known);
HOSTS_GENERATION.fetch_add(1, Ordering::Relaxed);
let mut shard = recorder.shards[0].lock().unwrap_or_else(|e| e.into_inner());
for ((hour, host), counts) in hours {
let host = map.get(host as usize).copied().unwrap_or(0);
let entry = shard.hours.entry((hour, host)).or_default();
for g in 0..3 {
entry[g].requests += counts[g].requests;
entry[g].pages += counts[g].pages;
}
}
for (day, lists) in days {
shard.days.entry(day).or_default().merge(&lists);
}
}
#[cfg(test)]
#[doc(hidden)]
pub fn disable_for_tests() {
ENABLED.store(false, Ordering::Relaxed);
}
#[cfg(test)]
#[doc(hidden)]
pub fn reset_for_tests() {
let recorder = recorder();
HOSTS_GENERATION.fetch_add(1, Ordering::Relaxed);
for local in locals().lock().unwrap_or_else(|e| e.into_inner()).iter().filter_map(|w| w.upgrade()) {
*local.lock().unwrap_or_else(|e| e.into_inner()) = Local::default();
}
for shard in &recorder.shards {
*shard.lock().unwrap_or_else(|e| e.into_inner()) = Shard::default();
}
recorder.hosts.write().unwrap_or_else(|e| e.into_inner()).clear();
}