use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use crate::error::Error;
use crate::geo::{SharedGeo, locate};
use crate::hash::{new_id, random_salt, visitor_hash};
use crate::http::{FetchInit, Headers, Request, Response, SharedFetcher, Url, default_fetcher};
use crate::js::{self, Object, Value};
use crate::limit::{RateLimit, Turns};
use crate::mail::Message;
use crate::mail::transports::{SERVICES, Service};
use crate::payload::{MAX_BODY, Payload, parse_payload};
use crate::re::{js_re, test};
use crate::safefetch::{Lookup, public_address};
use crate::sources::{Page, attribute, decode_uri_component, parse_page, strip_www};
use crate::store::{EVENT_TAIL_MS, EventRow, SessionRow, SiteOverrides, SiteRow, SqlStore};
use crate::time::{add_days, is_timezone, local_date, start_of};
use crate::ua::{ClientHints, ai_agent, is_bot, parse_client};
use crate::{BoxError, BoxFuture};
pub const LINK_DOMAIN_CHECK: &str = "/.well-known/runlight-link-domain";
const ROLLUP_VERSION: &str = "3";
const ROLLUP_BATCH: usize = 10;
pub(crate) const REMOTE_MAX_BYTES: usize = 2 * 1024 * 1024;
const METERED_ROLLUP_BATCH: usize = 4;
const ROLLUP_DELAY_MS: i64 = 2 * 3_600_000;
const TRACKER_WAIT: std::time::Duration = std::time::Duration::from_secs(10);
pub const RETENTION_MONTHS: [i64; 5] = [6, 12, 24, 36, 60];
pub const SESSION_IDLE_MS: i64 = 30 * 60 * 1000;
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct SiteOptions {
pub id: Option<String>,
pub name: Option<String>,
pub hostnames: Option<Vec<String>>,
pub timezone: Option<String>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub enum TrustProxy {
#[default]
On,
Off,
Header(String),
}
pub type Clock = Arc<dyn Fn() -> i64 + Send + Sync>;
pub type AppSend = Arc<dyn Fn(Message) -> BoxFuture<'static, Result<(), BoxError>> + Send + Sync>;
#[derive(Clone)]
pub struct AppMailer {
pub send: AppSend,
pub from: String,
pub from_name: Option<String>,
pub via: String,
}
impl std::fmt::Debug for AppMailer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AppMailer")
.field("from", &self.from)
.field("from_name", &self.from_name)
.field("via", &self.via)
.finish_non_exhaustive()
}
}
#[derive(Clone)]
pub struct RunlightOptions {
pub store: SqlStore,
pub site: Option<SiteOptions>,
pub sites: Option<Vec<SiteOptions>>,
pub managed_sites: bool,
pub geo: Option<SharedGeo>,
pub trust_proxy: Option<TrustProxy>,
pub link_path: Option<String>,
pub mail: Option<Object>,
pub app_mailer: Option<AppMailer>,
pub mail_services: Option<Vec<String>>,
pub secret: Option<String>,
pub rate_limit: Option<f64>,
pub now: Option<Clock>,
pub fetcher: Option<SharedFetcher>,
pub local_installs: bool,
pub lookup: Option<Lookup>,
}
impl RunlightOptions {
pub fn new(store: SqlStore) -> RunlightOptions {
RunlightOptions {
store,
site: None,
sites: None,
managed_sites: false,
geo: None,
trust_proxy: None,
link_path: None,
mail: None,
app_mailer: None,
mail_services: None,
secret: None,
rate_limit: None,
now: None,
fetcher: None,
local_installs: false,
lookup: None,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Remote {
pub url: String,
pub token: String,
pub site: String,
pub hostnames: Vec<String>,
pub scope: Option<String>,
}
impl Remote {
fn to_value(&self) -> Value {
let mut o = Object::new();
o.set("url", self.url.clone());
o.set("token", self.token.clone());
o.set("site", self.site.clone());
o.set("hostnames", Value::Array(self.hostnames.iter().map(|h| Value::from(h.as_str())).collect()));
if let Some(scope) = &self.scope {
o.set("scope", scope.clone());
}
Value::Object(o)
}
fn from_value(v: &Value) -> Remote {
Remote {
url: js::str_or_empty(v.get("url")),
token: js::str_or_empty(v.get("token")),
site: js::str_or_empty(v.get("site")),
hostnames: v
.get("hostnames")
.and_then(Value::as_array)
.map(|a| a.iter().map(js::js_string).collect())
.unwrap_or_default(),
scope: v.get("scope").and_then(Value::as_str).map(str::to_string),
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct RemoteInfo {
pub last_seen: Option<f64>,
pub retention_months: Option<Option<f64>>,
pub connection: &'static str,
}
pub fn env_value(name: &str) -> Option<String> {
std::env::var(name).ok().map(|v| js::trim(&v).to_string()).filter(|v| !v.is_empty())
}
pub fn is_email(text: &str) -> bool {
crate::re::uni_re!(r#"^[^@<>"]+@[^@<>"]+\.[^@<>"]+$"#).is_match(text) && !text.chars().any(js::is_space)
}
fn site_row(options: &SiteOptions, index: usize) -> Result<SiteRow, String> {
let timezone = options.timezone.clone().unwrap_or_else(|| "UTC".into());
if !is_timezone(&timezone) {
return Err(format!("Runlight: unknown timezone \"{timezone}\""));
}
let id = options.id.clone().unwrap_or_else(|| if index == 0 { "default".into() } else { String::new() });
if id.is_empty() || !test(js_re!(r"(?i)^[a-z0-9][a-z0-9._-]{0,63}$"), &id) {
return Err(format!("Runlight: site id \"{id}\" must be letters, digits, dots, dashes, or underscores"));
}
let hostnames = options.hostnames.clone().unwrap_or_default();
Ok(SiteRow {
id,
name: options.name.clone().unwrap_or_else(|| hostnames.first().cloned().unwrap_or_else(|| "My site".into())),
hostnames: hostnames.iter().map(|h| strip_www(h)).collect(),
timezone,
})
}
fn utc_day(ts: i64) -> String {
js::head16(&js::iso_string(ts), 10)
}
pub(crate) fn locale_compare(a: &str, b: &str) -> std::cmp::Ordering {
a.to_lowercase().cmp(&b.to_lowercase()).then_with(|| b.cmp(a))
}
#[derive(Default)]
struct State {
configured: Vec<SiteRow>,
overrides: HashMap<String, SiteOverrides>,
remotes: Vec<(String, Remote)>,
remote_seen: HashMap<String, (i64, RemoteInfo)>,
link_domains: Option<(i64, HashSet<String>)>,
salts: HashMap<String, (String, String, Option<String>)>,
optimized_at: i64,
}
pub(crate) struct Inner {
pub(crate) store: SqlStore,
pub(crate) managed_sites: bool,
geo: Option<SharedGeo>,
trust_proxy: TrustProxy,
warn_direct: AtomicBool,
direct_warnings: AtomicU32,
limit: Option<RateLimit>,
clock: Clock,
pub(crate) link_path: String,
pub(crate) mail_in_code: Option<Object>,
pub(crate) app_mailer: Option<AppMailer>,
pub(crate) mail_services: Option<Vec<&'static Service>>,
pub(crate) secret: Option<String>,
pub(crate) fetcher: SharedFetcher,
local_installs: bool,
lookup: Option<Lookup>,
state: Mutex<State>,
ready: tokio::sync::Mutex<bool>,
initialized: AtomicBool,
checking: tokio::sync::Mutex<()>,
turns: Turns,
pruning: Mutex<Vec<tokio::task::JoinHandle<()>>>,
prune_turn: Arc<tokio::sync::Mutex<()>>,
pub(crate) route_bases: Mutex<Vec<String>>,
}
#[derive(Clone)]
pub struct Runlight(pub(crate) Arc<Inner>);
impl Runlight {
pub fn new(options: RunlightOptions) -> Result<Runlight, String> {
let configured: Vec<SiteOptions> = if options.managed_sites {
vec![]
} else {
match &options.sites {
Some(sites) if !sites.is_empty() => sites.clone(),
_ => vec![options.site.clone().unwrap_or_default()],
}
};
let configured = configured.iter().enumerate().map(|(i, s)| site_row(s, i)).collect::<Result<Vec<_>, _>>()?;
if configured.len() > 1 && configured.iter().any(|s| s.hostnames.is_empty()) {
return Err("Runlight: with several sites, give each one its hostnames".into());
}
let ids: HashSet<&str> = configured.iter().map(|s| s.id.as_str()).collect();
if ids.len() != configured.len() {
return Err("Runlight: two sites share an id".into());
}
if options.mail.is_some() && options.app_mailer.is_some() {
return Err("Runlight: pass mail or app_mailer, not both".into());
}
let app_mailer = options.app_mailer.map(|mut app| {
app.from_name = app.from_name.filter(|n| !n.is_empty());
app
});
if app_mailer.as_ref().is_some_and(|app| !is_email(&app.from)) {
return Err("Runlight: the app's mailer needs a from address, like reports@example.com".into());
}
let mail_in_code = options.mail;
if let Some(id) = options.mail_services.iter().flatten().find(|id| !SERVICES.iter().any(|s| s.id == *id)) {
return Err(format!("Runlight: mail_services names \"{id}\", which is not a mail service"));
}
let mail_services = options
.mail_services
.as_ref()
.map(|offered| SERVICES.iter().filter(|s| offered.iter().any(|id| id == s.id)).collect());
let per_minute = options.rate_limit.unwrap_or(120.0);
let limit = (per_minute > 0.0).then(|| RateLimit::new(per_minute));
let clock: Clock = options.now.clone().unwrap_or_else(|| {
Arc::new(|| {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as i64)
})
});
let link_path = format!("/{}", options.link_path.as_deref().unwrap_or("/go").trim_matches('/'));
let secret =
options.secret.clone().or_else(|| env_value("RUNLIGHT_SECRET")).or_else(|| env_value("RUNLIGHT_TOKEN"));
Ok(Runlight(Arc::new(Inner {
store: options.store,
managed_sites: options.managed_sites,
geo: options.geo,
warn_direct: AtomicBool::new(options.trust_proxy.is_none()),
direct_warnings: AtomicU32::new(0),
trust_proxy: options.trust_proxy.unwrap_or_default(),
limit,
clock,
link_path,
mail_in_code,
app_mailer,
mail_services,
secret,
fetcher: options.fetcher.unwrap_or_else(default_fetcher),
local_installs: options.local_installs,
lookup: options.lookup,
state: Mutex::new(State { configured, ..State::default() }),
ready: tokio::sync::Mutex::new(false),
initialized: AtomicBool::new(false),
checking: tokio::sync::Mutex::new(()),
turns: Turns::default(),
pruning: Mutex::new(Vec::new()),
prune_turn: Arc::new(tokio::sync::Mutex::new(())),
route_bases: Mutex::new(Vec::new()),
})))
}
fn state(&self) -> std::sync::MutexGuard<'_, State> {
self.0.state.lock().unwrap_or_else(|e| e.into_inner())
}
pub fn app_mailer(&self) -> Option<&AppMailer> {
self.0.app_mailer.as_ref()
}
pub fn mail_services(&self) -> Vec<&'static Service> {
self.0.mail_services.clone().unwrap_or_else(|| SERVICES.iter().collect())
}
pub fn store(&self) -> &SqlStore {
&self.0.store
}
pub fn now(&self) -> i64 {
(self.0.clock)()
}
pub fn fetcher(&self) -> &SharedFetcher {
&self.0.fetcher
}
pub fn local_installs(&self) -> bool {
self.0.local_installs
}
pub(crate) fn lookup(&self) -> Option<&Lookup> {
self.0.lookup.as_ref()
}
pub(crate) async fn fetch_install(&self, url: &str, init: FetchInit) -> Result<Response, crate::http::FetchError> {
crate::safefetch::install_fetch(&*self.0.fetcher, url, init, self.0.local_installs, self.lookup()).await
}
pub fn managed_sites(&self) -> bool {
self.0.managed_sites
}
pub fn secret(&self) -> Option<&str> {
self.0.secret.as_deref()
}
pub fn link_path(&self) -> &str {
&self.0.link_path
}
pub async fn init(&self) -> Result<(), Error> {
let mut ready = self.0.ready.lock().await;
if *ready {
return Ok(());
}
let store = &self.0.store;
store.migrate().await?;
store.optimize(true).await;
if self.0.managed_sites {
let sites = store.sites().await?;
self.state().configured = sites;
self.load_remotes().await?;
}
let configured = self.state().configured.clone();
for site in &configured {
store.upsert_site(site, self.now()).await?;
}
let overrides = store.site_overrides().await?;
self.state().overrides = overrides;
for site in self.sites() {
if self.is_remote(&site.id) {
continue;
}
let stored = store.setting(&format!("rollup-zone:{}", site.id)).await?;
let zone = stored.and_then(|s| js::parse(&s).ok()).map(|v| js::str_or_empty(v.get("zone")));
match zone {
None => {
let value = obj_zone(&site.timezone, 0);
store.set_setting(&format!("rollup-zone:{}", site.id), Some(&value)).await?;
}
Some(zone) if zone != site.timezone => {
self.zone_changed(&site.id, &site.timezone).await?;
}
_ => {}
}
}
*ready = true;
self.0.initialized.store(true, Ordering::Release);
Ok(())
}
async fn ready_within(&self, wait: std::time::Duration) -> Result<bool, Error> {
if self.0.initialized.load(Ordering::Acquire) {
return Ok(true);
}
let rl = self.clone();
let init = tokio::spawn(async move { rl.init().await });
match tokio::time::timeout(wait, init).await {
Ok(Ok(done)) => done.map(|()| true),
Ok(Err(stopped)) => Err(Error::Other(format!("Runlight: {stopped}"))),
Err(_) => Ok(false),
}
}
pub fn sites(&self) -> Vec<SiteRow> {
let state = self.state();
state
.configured
.iter()
.map(|site| {
let mut site = site.clone();
if let Some(o) = state.overrides.get(&site.id) {
if let Some(Value::String(name)) = o.get("name") {
site.name = name.clone();
}
if let Some(Value::String(tz)) = o.get("timezone") {
site.timezone = tz.clone();
}
}
site
})
.collect()
}
pub(crate) fn configured(&self) -> Vec<SiteRow> {
self.state().configured.clone()
}
fn hostnames_for(&self, input: Option<&Value>, except: Option<&str>) -> Result<Vec<String>, Error> {
let list: Vec<String> = match input {
Some(Value::Array(a)) => a.iter().map(js::js_string).collect(),
other => {
let text = js::str_or_empty(other);
crate::re::uni_re!(r"[\t\n\x{0B}\f\r \x{A0}\x{1680}\x{2000}-\x{200A}\x{2028}\x{2029}\x{202F}\x{205F}\x{3000}\x{FEFF},]+")
.split(&text)
.map(str::to_string)
.collect()
}
};
let mut hostnames: Vec<String> = Vec::new();
for h in list {
let h = js::trim(&h).to_string();
let h = crate::re::replace_first(js_re!(r"^https?://"), &h, "");
let h = crate::re::replace_first(js_re!(r"[/:][\s\S]*$"), &h, "");
let h = strip_www(&h);
if !h.is_empty() && !hostnames.contains(&h) {
hostnames.push(h);
}
}
if hostnames.is_empty() {
return Err(Error::settings("Add the site's domain, like example.com", "site_domain_needed", &[]));
}
let configured = self.configured();
for host in &hostnames {
if !is_domain(host) && host != "localhost" {
return Err(Error::settings(
format!("\"{host}\" is not a domain name"),
"site_domain_invalid",
&[("host", host)],
));
}
if let Some(owner) = configured.iter().find(|s| Some(s.id.as_str()) != except && s.hostnames.contains(host))
{
return Err(Error::settings(
format!("{host} already belongs to {}", owner.name),
"site_domain_taken",
&[("host", host), ("site", &owner.name)],
));
}
}
Ok(hostnames)
}
async fn load_remotes(&self) -> Result<(), Error> {
let mut remotes = Vec::new();
for (key, value) in self.0.store.settings_starting_with("remote:").await? {
if let Some(opened) = crate::mail::secret::unseal(&value, self.0.secret.as_deref())
&& let Ok(v) = js::parse(&opened)
{
remotes.push((key["remote:".len()..].to_string(), Remote::from_value(&v)));
}
}
self.state().remotes = remotes;
Ok(())
}
pub fn is_remote(&self, id: &str) -> bool {
self.state().remotes.iter().any(|(k, _)| k == id)
}
pub fn remote(&self, id: &str) -> Option<Remote> {
self.state().remotes.iter().find(|(k, _)| k == id).map(|(_, r)| r.clone())
}
pub async fn remote_last_seen(&self, id: &str) -> Option<f64> {
self.remote_info(id).await.and_then(|i| i.last_seen)
}
pub async fn remote_info(&self, id: &str) -> Option<RemoteInfo> {
let remote = self.remote(id)?;
let cached = self.state().remote_seen.get(id).cloned();
if let Some((at, info)) = &cached
&& self.now() - at < 60_000
{
return Some(info.clone());
}
let mut info = RemoteInfo {
last_seen: cached.as_ref().and_then(|c| c.1.last_seen),
retention_months: None,
connection: "unreachable",
};
let init = FetchInit::default()
.header("authorization", format!("Bearer {}", remote.token))
.timeout(8000)
.max_bytes(REMOTE_MAX_BYTES);
if let Ok(answer) = self.fetch_install(&format!("{}/api/sites", remote.url), init).await {
if answer.status == 401 || answer.status == 403 {
info.connection = "refused";
}
if let Ok(body) = answer.json_body()
&& let Some(there) = body.get("sites").and_then(Value::as_array).and_then(|a| {
a.iter().find(|s| {
js::str_or_empty(s.get("id")) == remote.site && s.get("id").is_some_and(Value::is_string)
})
})
{
info = RemoteInfo {
last_seen: there.get("lastSeen").and_then(Value::as_f64),
retention_months: Some(there.get("retentionMonths").and_then(Value::as_f64)),
connection: "ok",
};
}
}
self.state().remote_seen.insert(id.to_string(), (self.now(), info.clone()));
Some(info)
}
pub fn forget_remote_info(&self, id: &str) {
self.state().remote_seen.remove(id);
}
async fn revoke_remote_token(&self, remote: &Remote) {
let init =
FetchInit::method("DELETE").header("authorization", format!("Bearer {}", remote.token)).timeout(5000);
let _ = self.fetch_install(&format!("{}/api/token", remote.url), init).await;
}
async fn seal(&self, value: &str) -> String {
crate::mail::secret::seal(value, self.0.secret.as_deref())
}
async fn add_remote_site(&self, input: &Value, name: Option<&Value>) -> Result<SiteRow, Error> {
let url = js::trim(&js::str_or_empty(input.get("url"))).trim_end_matches('/').to_string();
if !crate::safefetch::install_address(&url, self.0.local_installs) {
return Err(Error::settings(
"Enter the install's address, like https://example.com/runlight",
"connect_url",
&[],
));
}
let token = js::trim(&js::str_or_empty(input.get("token"))).to_string();
if token.is_empty() {
return Err(Error::settings("Enter an API token from that install", "install_token", &[]));
}
let auth = format!("Bearer {token}");
let answer = match self
.fetch_install(
&format!("{url}/api/sites"),
FetchInit::default().header("authorization", &auth).timeout(10_000).max_bytes(REMOTE_MAX_BYTES),
)
.await
{
Ok(a) => a,
Err(crate::http::FetchError::TooLong(_)) => Response::status(200),
Err(_) => {
let host = Url::parse(&url).map(|u| u.host()).unwrap_or_default();
return Err(Error::settings(format!("Could not reach {url}"), "unreachable", &[("host", &host)]));
}
};
if answer.status == 401 || answer.status == 403 {
return Err(Error::settings("That install refused the token", "install_refused", &[]));
}
let body = answer.json_body().ok();
let sites: Vec<Value> =
body.as_ref().and_then(|b| b.get("sites")).and_then(Value::as_array).cloned().unwrap_or_default();
if !answer.ok() || sites.is_empty() {
return Err(Error::settings(
format!("{url} did not answer like a Runlight install"),
"connect_not_runlight",
&[("url", &url)],
));
}
let mut scope = "read".to_string();
let mut token_site = String::new();
if let Ok(about) = self
.fetch_install(
&format!("{url}/api/token"),
FetchInit::default().header("authorization", &auth).timeout(10_000).max_bytes(REMOTE_MAX_BYTES),
)
.await
&& about.ok()
&& let Ok(info) = about.json_body()
{
if info.get("scope").and_then(Value::as_str) == Some("manage") {
scope = "manage".into();
}
token_site = js::str_or_empty(info.get("site"));
}
let wanted = if token_site.is_empty() { input.get("site").cloned() } else { Some(Value::String(token_site)) };
let there = sites.iter().find(|s| match (&wanted, s.get("id")) {
(Some(w), Some(id)) => w == id,
(None, None) => true,
_ => false,
});
let there = there.unwrap_or(&sites[0]).clone();
let hostnames: Vec<String> = there
.get("hostnames")
.and_then(Value::as_array)
.map(|a| a.iter().filter_map(|h| h.as_str().map(str::to_string)).collect())
.unwrap_or_default();
let there_id = js::str_or_empty(there.get("id"));
let known: Vec<(String, Remote)> = self.state().remotes.clone();
for (existing, k) in known {
if k.url == url && k.site == there_id {
let updated = Remote {
token: token.clone(),
scope: Some(scope.clone()),
hostnames: hostnames.clone(),
..k.clone()
};
if k.token != token {
self.revoke_remote_token(&k).await;
}
let sealed = self.seal(&updated.to_value().to_json()).await;
self.0.store.set_setting(&format!("remote:{existing}"), Some(&sealed)).await?;
{
let mut state = self.state();
if let Some(slot) = state.remotes.iter_mut().find(|(id, _)| *id == existing) {
slot.1 = updated;
}
state.remote_seen.remove(&existing);
}
return self.site(Some(&existing)).ok_or_else(|| Error::Range("Unknown site".into()));
}
}
let host_source =
hostnames.first().cloned().unwrap_or_else(|| Url::parse(&url).map(|u| u.host()).unwrap_or_default());
let host = crate::re::replace_all(js_re!(r"(?i)[^a-z0-9._-]"), &host_source, "-").to_lowercase();
let stem = js::head16(&host, 56);
let mut id = stem.clone();
let mut n = 2;
while self.configured().iter().any(|s| s.id == id) {
id = format!("{stem}-{n}");
n += 1;
}
let given = js::head16(js::trim(&js::str_or_empty(name)), 80);
let site_name = if given.is_empty() { js::str_or_empty(there.get("name")) } else { given };
let tz = js::str_or_empty(there.get("timezone"));
let site = SiteRow {
id: id.clone(),
name: site_name,
hostnames: vec![],
timezone: if is_timezone(&tz) { tz } else { "UTC".into() },
};
let remote = Remote { url, token, site: there_id, hostnames, scope: Some(scope) };
self.0.store.upsert_site(&site, self.now()).await?;
let sealed = self.seal(&remote.to_value().to_json()).await;
self.0.store.set_setting(&format!("remote:{id}"), Some(&sealed)).await?;
let mut state = self.state();
state.remotes.push((id, remote));
state.configured.push(site.clone());
state.configured.sort_by(|a, b| locale_compare(&a.name, &b.name));
Ok(site)
}
pub async fn add_site(&self, input: &Value) -> Result<SiteRow, Error> {
self.init().await?;
if !self.0.managed_sites {
return Err(Error::settings("Sites are set in code", "sites_in_code", &[]));
}
if let Some(remote) = input.get("remote")
&& js::truthy(remote)
&& (remote.is_object() || remote.is_array())
{
return self.add_remote_site(remote, input.get("name")).await;
}
let hostnames = self.hostnames_for(input.get("hostnames"), None)?;
let given = js::trim(&js::str_or_empty(input.get("name"))).to_string();
let name = if given.is_empty() { hostnames[0].clone() } else { given };
if js::len16(&name) > 80 {
return Err(Error::settings("A site name is 1 to 80 characters", "site_name", &[]));
}
let timezone = input.get("timezone").filter(|v| !v.is_null()).map_or_else(|| "UTC".to_string(), js::js_string);
if !is_timezone(&timezone) {
return Err(Error::settings(
format!("Unknown timezone \"{timezone}\""),
"unknown_timezone",
&[("timezone", &timezone)],
));
}
let stem = js::head16(&crate::re::replace_all(js_re!(r"[^a-z0-9._-]"), &hostnames[0], "-"), 56);
let mut id = stem.clone();
let mut n = 2;
while self.configured().iter().any(|s| s.id == id) {
id = format!("{stem}-{n}");
n += 1;
}
let site = SiteRow { id, name, hostnames, timezone };
self.0.store.upsert_site(&site, self.now()).await?;
let mut state = self.state();
state.configured.push(site.clone());
state.configured.sort_by(|a, b| locale_compare(&a.name, &b.name));
Ok(site)
}
pub async fn delete_site(&self, id: &str) -> Result<(), Error> {
self.init().await?;
if !self.0.managed_sites {
return Err(Error::settings("Sites are set in code", "sites_in_code", &[]));
}
if !self.configured().iter().any(|s| s.id == id) {
return Err(Error::settings("Unknown site", "unknown_site", &[]));
}
let store = &self.0.store;
store.delete_site(id).await?;
for key in ["retention", "observe-key", "rollup-zone", "orphans-swept"] {
store.set_setting(&format!("{key}:{id}"), None).await?;
}
for (key, _) in store.settings_starting_with(&format!("import:umami-visits:{id}:")).await? {
store.set_setting(&key, None).await?;
}
if let Some(remote) = self.remote(id) {
self.revoke_remote_token(&remote).await;
self.state().remotes.retain(|(k, _)| k != id);
store.set_setting(&format!("remote:{id}"), None).await?;
}
let mut state = self.state();
state.configured.retain(|s| s.id != id);
state.overrides.remove(id);
Ok(())
}
pub async fn update_site(&self, id: &str, patch: &Value) -> Result<SiteRow, Error> {
self.init().await?;
let Some(current) = self.configured().into_iter().find(|s| s.id == id) else {
return Err(Error::settings("Unknown site", "unknown_site", &[]));
};
let name_of = |v: &Value| -> Result<String, Error> {
let name = js::trim(&js::js_string(v)).to_string();
if name.is_empty() || js::len16(&name) > 80 {
return Err(Error::settings("A site name is 1 to 80 characters", "site_name", &[]));
}
Ok(name)
};
let zone_of = |v: &Value| -> Result<String, Error> {
let tz = js::js_string(v);
if !is_timezone(&tz) {
return Err(Error::settings(
format!("Unknown timezone \"{tz}\""),
"unknown_timezone",
&[("timezone", &tz)],
));
}
Ok(tz)
};
let name = patch.get("name").map(name_of).transpose()?;
let timezone = patch.get("timezone").map(zone_of).transpose()?;
let hostnames = match patch.get("hostnames") {
Some(hostnames) if self.0.managed_sites && !self.is_remote(id) => {
Some(self.hostnames_for(Some(hostnames), Some(id))?)
}
_ => None,
};
let retention = match patch.get("retentionMonths") {
Some(r) => {
let months = if r.is_null() { None } else { Some(js::js_number(r)) };
let months = months.map(|n| if n.fract() == 0.0 { n as i64 } else { -1 });
self.check_retention(id, months)?;
Some(months)
}
None => None,
};
if let Some(tz) = &timezone
&& Some(tz.clone()) != self.site(Some(id)).map(|s| s.timezone)
{
self.zone_changed(id, tz).await?;
}
if self.0.managed_sites {
let mut next = current.clone();
if let Some(name) = name {
next.name = name;
}
if let Some(tz) = timezone {
next.timezone = tz;
}
if let Some(hostnames) = hostnames {
next.hostnames = hostnames;
}
self.0.store.upsert_site(&next, self.now()).await?;
let mut state = self.state();
if let Some(slot) = state.configured.iter_mut().find(|s| s.id == id) {
*slot = next;
}
} else {
let mut next = self.state().overrides.get(id).cloned().unwrap_or_default();
if let Some(name) = name {
next.set("name", name);
}
if let Some(tz) = timezone {
next.set("timezone", tz);
}
self.0.store.set_site_overrides(id, &next).await?;
self.state().overrides.insert(id.to_string(), next);
}
if let Some(months) = retention {
self.set_retention(id, months).await?;
}
self.site(Some(id)).ok_or_else(|| Error::Range("Unknown site".into()))
}
pub async fn retention(&self, site: &str) -> Result<Option<i64>, Error> {
let value = self.0.store.setting(&format!("retention:{site}")).await?;
let n = value.map_or(0.0, |v| js::text_number(&v));
Ok(RETENTION_MONTHS.iter().find(|m| **m as f64 == n).copied())
}
pub async fn set_retention(&self, site: &str, months: Option<i64>) -> Result<(), Error> {
self.check_retention(site, months)?;
self.0.store.set_setting(&format!("retention:{site}"), months.map(|m| m.to_string()).as_deref()).await?;
self.prune_later(Some(site.to_string()));
Ok(())
}
fn check_retention(&self, site: &str, months: Option<i64>) -> Result<(), Error> {
if self.site(Some(site)).is_none() || self.is_remote(site) {
return Err(Error::settings("Unknown site", "unknown_site", &[]));
}
if let Some(m) = months
&& !RETENTION_MONTHS.contains(&m)
{
return Err(Error::settings(
"Keep visits for 6, 12, 24, 36, 60 months, or forever",
"retention_bad",
&[("months", "6, 12, 24, 36, 60")],
));
}
Ok(())
}
fn prune_later(&self, only: Option<String>) {
let rl = self.clone();
let turn = self.0.prune_turn.clone();
let handle = tokio::spawn(async move {
let _turn = turn.lock().await;
if let Err(error) = rl.apply_retention(only.as_deref()).await {
eprintln!("Runlight: could not apply retention {error}");
}
});
self.0.pruning.lock().unwrap_or_else(|e| e.into_inner()).push(handle);
}
pub fn later<F>(&self, work: F)
where
F: std::future::Future<Output = ()> + Send + 'static,
{
let handle = tokio::spawn(work);
self.0.pruning.lock().unwrap_or_else(|e| e.into_inner()).push(handle);
}
pub async fn idle(&self) {
loop {
let handles: Vec<_> = std::mem::take(&mut *self.0.pruning.lock().unwrap_or_else(|e| e.into_inner()));
if handles.is_empty() {
return;
}
for h in handles {
let _ = h.await;
}
}
}
async fn zone_changed(&self, id: &str, timezone: &str) -> Result<i64, Error> {
let since = self.now();
self.0.store.clear_rollups(id, None, None).await?;
self.0.store.set_setting(&format!("rollup-zone:{id}"), Some(&obj_zone(timezone, since))).await?;
Ok(since)
}
async fn rollup_since(&self, site: &SiteRow) -> Result<Option<i64>, Error> {
let key = format!("rollup-zone:{}", site.id);
let Some(stored) = self.0.store.setting(&key).await? else {
self.0.store.set_setting(&key, Some(&obj_zone(&site.timezone, 0))).await?;
return Ok(Some(0));
};
let zone = js::parse(&stored).unwrap_or(Value::Null);
Ok((js::str_or_empty(zone.get("zone")) == site.timezone).then(|| js::to_i64(js::opt_number(zone.get("since")))))
}
pub async fn build_rollups(&self) -> Result<usize, Error> {
let store = &self.0.store;
if store.setting("rollup-version").await?.as_deref() != Some(ROLLUP_VERSION) {
for site in self.sites() {
store.clear_rollups(&site.id, None, None).await?;
}
store.set_setting("rollup-version", Some(ROLLUP_VERSION)).await?;
}
let mut built = 0;
let now = self.now();
for site in self.sites() {
if self.is_remote(&site.id) {
continue;
}
let Some(first) = store.first_seen(&site.id).await? else { continue };
let cutoff = self.retention_cutoff(&site.id).await?.unwrap_or(0);
let Some(since) = self.rollup_since(&site).await? else { continue };
let done = store.rollup_days(&site.id).await?;
let today = local_date(now, &site.timezone);
let batch = if store.db().metered() { METERED_ROLLUP_BATCH } else { ROLLUP_BATCH };
let oldest = local_date(first.max(cutoff), &site.timezone);
let mut made = 0;
let mut day = add_days(&today, -1);
while day >= oldest && made < batch {
if !done.contains(&day) {
let start = start_of(&day, &site.timezone, 0);
let end = start_of(&add_days(&day, 1), &site.timezone, 0);
if start < since {
break;
}
if !(now < end + ROLLUP_DELAY_MS || start < cutoff) {
match store.build_rollup_day(&site.id, &day, start, end).await {
Ok(()) => made += 1,
Err(error) => {
if !store.rollup_days(&site.id).await?.contains(&day) {
eprintln!("Runlight: could not add up {day} for {} {error}", site.id);
}
}
}
tokio::task::yield_now().await;
}
}
day = add_days(&day, -1);
}
built += made;
}
Ok(built)
}
pub async fn retention_cutoff(&self, site: &str) -> Result<Option<i64>, Error> {
let Some(months) = self.retention(site).await? else { return Ok(None) };
Ok(Some(minus_months(self.now(), months)))
}
async fn apply_retention(&self, only: Option<&str>) -> Result<(), Error> {
let store = &self.0.store;
for site in self.sites() {
if only.is_some_and(|o| o != site.id) || self.is_remote(&site.id) {
continue;
}
let Some(cutoff) = self.retention_cutoff(&site.id).await? else { continue };
store.drop_before(&site.id, cutoff).await?;
let key = format!("orphans-swept:{}", site.id);
if store.setting(&key).await?.is_none_or(|v| v.is_empty()) {
store.drop_orphans(&site.id, cutoff, self.now()).await?;
store.set_setting(&key, Some("1")).await?;
}
}
Ok(())
}
pub fn site(&self, id: Option<&str>) -> Option<SiteRow> {
let sites = self.sites();
match id.filter(|i| !i.is_empty()) {
None => sites.into_iter().next(),
Some(id) => sites.into_iter().find(|s| s.id == id),
}
}
pub fn site_for(&self, hostname: &str, id: Option<&str>) -> Option<SiteRow> {
let host = strip_www(hostname);
let id = id.filter(|i| !i.is_empty());
let remotes: Vec<String> = self.state().remotes.iter().map(|(k, _)| k.clone()).collect();
let sites = self.sites();
if !remotes.is_empty() {
let local: Vec<SiteRow> = sites.into_iter().filter(|s| !remotes.contains(&s.id)).collect();
if let Some(id) = id
&& remotes.iter().any(|r| r == id)
{
return None;
}
return site_among(&local, &host, id);
}
site_among(&sites, &host, id)
}
async fn setup_site(&self, hostname: &str, id: Option<&str>) -> Result<Option<SiteRow>, Error> {
let lower = hostname.to_lowercase();
let host = lower.trim_start_matches('[').trim_end_matches(']');
let host = {
let mut h = lower.as_str();
if let Some(rest) = h.strip_prefix('[') {
h = rest;
}
if let Some(rest) = h.strip_suffix(']') {
h = rest;
}
let _ = host;
h.to_string()
};
if !(host == "localhost"
|| host == "127.0.0.1"
|| host == "::1"
|| test(js_re!(r"\.(localhost|local|test)$"), &host))
{
return Ok(None);
}
let site = match id.filter(|i| !i.is_empty()) {
Some(id) => self.site(Some(id)),
None => {
let sites = self.sites();
if sites.len() == 1 { sites.into_iter().next() } else { None }
}
};
let Some(site) = site else { return Ok(None) };
if self.is_remote(&site.id) {
return Ok(None);
}
Ok(self.0.store.last_seen(&site.id).await?.is_none().then_some(site))
}
pub fn client_ip(&self, request: &Request) -> String {
let h = &request.headers;
let last = |name: &str| -> Option<String> {
let value = h.get(name)?;
value.split(',').map(|x| js::trim(x).to_string()).rfind(|x| !x.is_empty())
};
let forwarded = match &self.0.trust_proxy {
TrustProxy::Off => None,
TrustProxy::On => {
last("x-forwarded-for").or_else(|| h.get("x-real-ip")).or_else(|| h.get("cf-connecting-ip"))
}
TrustProxy::Header(name) if name == "x-forwarded-for" => last("x-forwarded-for"),
TrustProxy::Header(name) => h.get(name),
};
if let Some(f) = forwarded
&& !js::trim(&f).is_empty()
{
return js::trim(&f).to_string();
}
if self.0.warn_direct.load(Ordering::Relaxed)
&& public_address(&request.remote_address)
&& self.0.warn_direct.swap(false, Ordering::Relaxed)
{
self.0.direct_warnings.fetch_add(1, Ordering::Relaxed);
eprintln!(
"Runlight: a request came straight from a public address with no proxy in front, but trustProxy is on by default, so a client could send X-Forwarded-For and choose its own address, getting round the rate limits. Set trust_proxy: Some(TrustProxy::Off) when nothing sits in front of this server, or put a proxy in front that sets the header."
);
}
request.remote_address.clone()
}
#[doc(hidden)]
pub fn direct_warnings(&self) -> u32 {
self.0.direct_warnings.load(Ordering::Relaxed)
}
async fn current_salts(&self, now: i64, timezone: &str) -> Result<(String, Option<String>), Error> {
let day = local_date(now, timezone);
if let Some((d, today, yesterday)) = self.state().salts.get(timezone).cloned()
&& d == day
{
return Ok((today, yesterday));
}
let today = self.0.store.salt(&day, &random_salt()).await?;
let yesterday = self.0.store.salt_if_exists(&add_days(&day, -1)).await?;
self.drop_old_salts(now).await?;
self.state().salts.insert(timezone.to_string(), (day, today.clone(), yesterday.clone()));
Ok((today, yesterday))
}
async fn drop_old_salts(&self, now: i64) -> Result<(), Error> {
self.0.store.drop_salts_before(&utc_day(now - 2 * 86_400_000)).await?;
Ok(())
}
pub(crate) fn forwarded_host(&self, request: &Request) -> Option<String> {
if self.0.trust_proxy == TrustProxy::Off { None } else { request.headers.get("x-forwarded-host") }
}
pub async fn collect(&self, request: &Request) -> Result<(), Error> {
let length = request.headers.get("content-length").map_or(0.0, |l| js::text_number(&l));
if length > MAX_BODY as f64 {
return Ok(());
}
if request.body.len() > MAX_BODY {
return Ok(());
}
let text = request.text();
let Some(payload) = parse_payload(&text) else { return Ok(()) };
let ua = request.headers.get("user-agent").unwrap_or_default();
if ai_agent(&ua).is_some() || is_bot(&ua) {
return Ok(());
}
if let Some(limit) = &self.0.limit
&& !limit.allow(&self.client_ip(request), self.now())
{
return Ok(());
}
if !self.ready_within(TRACKER_WAIT).await? {
return Ok(());
}
let now = self.now();
let mut attempt = 1;
loop {
match self.record(&payload, request, now).await {
Err(error) if attempt < 3 && busy(&error) => {
tokio::time::sleep(std::time::Duration::from_millis(500 * attempt)).await;
attempt += 1;
}
other => return other,
}
}
}
async fn record(&self, payload: &Payload, request: &Request, now: i64) -> Result<(), Error> {
self.init().await?;
let hostname = payload.url.hostname();
let site = match self.site_for(&hostname, Some(&payload.site)) {
Some(s) => s,
None => match self.setup_site(&hostname, Some(&payload.site)).await? {
Some(s) => s,
None => return Ok(()),
},
};
if payload.kind == "engagement" {
return self.engagement(&site, payload, now).await;
}
let store = &self.0.store;
let page = parse_page(&payload.url);
let mut session: Option<(String, String)> = None;
let mut reopen = true;
if payload.kind == "event"
&& !payload.pageview_id.is_empty()
&& let Some(pageview) = store.pageview(&site.id, &payload.pageview_id).await?
&& now - pageview.started_at < EVENT_TAIL_MS
{
session = Some((pageview.session.clone(), pageview.visitor.clone()));
reopen = now - pageview.last_at <= SESSION_IDLE_MS;
if now - pageview.started_at > 3_600_000 {
store.touched_old_visit(&site.id, pageview.started_at, now - ROLLUP_DELAY_MS + 3_600_000).await?;
}
}
let (id, visitor) = match session {
Some(s) => s,
None => {
let screen = match (payload.screen_width, payload.screen_height) {
(Some(w), Some(h)) if w != 0 && h != 0 => format!("{w}x{h}"),
_ => String::new(),
};
self.session_for(
&site,
request,
&page,
&payload.referrer,
now,
payload.screen_width.map(|w| w as f64),
&screen,
&payload.language,
)
.await?
}
};
store.touch_session(&id, now, payload.kind, &page.path, reopen).await?;
store
.insert_event(&EventRow {
site: site.id.clone(),
ts: now,
kind: payload.kind.into(),
visitor,
session: id,
pageview: payload.pageview_id.clone(),
path: page.path.clone(),
hostname: page.hostname.clone(),
title: if payload.kind == "pageview" { payload.title.clone() } else { String::new() },
name: if payload.kind == "event" { payload.name.clone() } else { String::new() },
props: payload.props.clone(),
engaged_ms: 0,
scroll: None,
link: String::new(),
})
.await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn session_for(
&self,
site: &SiteRow,
request: &Request,
page: &Page,
referrer: &str,
now: i64,
screen_width: Option<f64>,
screen: &str,
language: &str,
) -> Result<(String, String), Error> {
let ua = request.headers.get("user-agent").unwrap_or_default();
let ip = self.client_ip(request);
let (today_salt, yesterday_salt) = self.current_salts(now, &site.timezone).await?;
let today = visitor_hash(&today_salt, &site.id, &ip, &ua);
let mut candidates = vec![today.clone()];
if let Some(y) = yesterday_salt {
candidates.push(visitor_hash(&y, &site.id, &ip, &ua));
}
let _turn = self.0.turns.take(&format!("{}:{today}", site.id)).await;
async {
let store = &self.0.store;
if let Some(open) = store.open_session(&site.id, &candidates, now - SESSION_IDLE_MS).await? {
return Ok(open);
}
let id = new_id();
let attribution = attribute(page, referrer, &site.hostnames);
let parsed = parse_client(
&ua,
&ClientHints {
brands: request.headers.get("sec-ch-ua"),
mobile: request.headers.get("sec-ch-ua-mobile"),
platform: request.headers.get("sec-ch-ua-platform"),
},
screen_width,
);
let location =
locate(&request.headers, &ip, self.0.geo.as_ref(), self.0.trust_proxy != TrustProxy::Off).await;
store
.insert_session(&SessionRow {
id: id.clone(),
site: site.id.clone(),
visitor: today.clone(),
started_at: now,
hostname: page.hostname.clone(),
referrer_host: attribution.referrer_host,
referrer_path: attribution.referrer_path,
source: attribution.source,
channel: attribution.channel.into(),
utm_source: page.utm.source.clone(),
utm_medium: page.utm.medium.clone(),
utm_campaign: page.utm.campaign.clone(),
utm_term: page.utm.term.clone(),
utm_content: page.utm.content.clone(),
country: location.country,
region: location.region,
city: location.city,
browser: parsed.browser,
browser_version: parsed.browser_version,
os: parsed.os,
os_version: parsed.os_version,
device: parsed.device.into(),
screen: screen.into(),
language: language.into(),
})
.await?;
Ok::<(String, String), Error>((id, today.clone()))
}
.await
}
async fn link_domain_set(&self) -> Result<HashSet<String>, Error> {
let now = self.now();
if let Some((at, domains)) = &self.state().link_domains
&& now - at < 30_000
{
return Ok(domains.clone());
}
self.init().await?;
let domains: HashSet<String> = self.0.store.link_domains().await?.into_iter().map(|(d, _)| d).collect();
self.state().link_domains = Some((now, domains.clone()));
Ok(domains)
}
pub fn forget_link_domains(&self) {
self.state().link_domains = None;
}
pub async fn link_response(&self, request: &Request) -> Result<Response, Error> {
let path = request.parsed_url().pathname();
let prefix = format!("{}/", self.0.link_path);
let slug = match path.strip_prefix(&prefix) {
Some(rest) => decode_uri_component(rest).unwrap_or_default(),
None => String::new(),
};
let found =
if !slug.is_empty() && !slug.contains('/') { self.redirect(request, &slug, "").await? } else { None };
Ok(found.unwrap_or_else(not_found))
}
pub async fn link_domain_response(&self, request: &Request) -> Result<Option<Response>, Error> {
let url = request.parsed_url();
let given = self.forwarded_host(request).or_else(|| request.headers.get("host")).unwrap_or_else(|| url.host());
let first = js::trim(given.split(',').next().unwrap_or("")).to_string();
let host = strip_www(first.split(':').next().unwrap_or(""));
if !self.link_domain_set().await?.contains(&host) {
return Ok(None);
}
let pathname = url.pathname();
if pathname == LINK_DOMAIN_CHECK {
return Ok(Some(Response::new(
crate::obj! { "runlight" => true, "domain" => host }.to_json(),
200,
Headers::new().with("content-type", "application/json").with("cache-control", "no-store"),
)));
}
let bases = {
let b = self.0.route_bases.lock().unwrap_or_else(|e| e.into_inner()).clone();
if b.is_empty() { vec!["/runlight".to_string()] } else { b }
};
for base in bases {
if base != "/" && (pathname == base || pathname.starts_with(&format!("{base}/"))) {
return Ok(None);
}
}
let slug = decode_uri_component(&pathname[1.min(pathname.len())..]).unwrap_or_default();
let found =
if !slug.is_empty() && !slug.contains('/') { self.redirect(request, &slug, &host).await? } else { None };
Ok(Some(found.unwrap_or_else(not_found)))
}
pub async fn redirect(&self, request: &Request, slug: &str, domain: &str) -> Result<Option<Response>, Error> {
self.init().await?;
let url = request.parsed_url();
let given = self.forwarded_host(request).or_else(|| request.headers.get("host")).unwrap_or_else(|| url.host());
let host = strip_www(given.split(':').next().unwrap_or(""));
let Some(link) = self.0.store.link_by_slug(slug).await? else { return Ok(None) };
if !domain.is_empty() && link.domain != domain {
return Ok(None);
}
let site = self.site(Some(&link.site)).or_else(|| self.sites().into_iter().next());
let ua = request.headers.get("user-agent").unwrap_or_default();
if let Some(site) = site
&& ai_agent(&ua).is_none()
&& !is_bot(&ua)
&& request.method == "GET"
{
let recorded = async {
let now = self.now();
let accept = request.headers.get("accept-language").unwrap_or_default();
let first = accept.split(',').next().unwrap_or("").split(';').next().unwrap_or("");
let language = js::head16(js::trim(first), 35);
let referer = request.headers.get("referer").unwrap_or_default();
let (id, visitor) =
self.session_for(&site, request, &parse_page(&url), &referer, now, None, "", &language).await?;
let pathname = url.pathname();
self.0.store.touch_session(&id, now, "click", &pathname, true).await?;
self.0
.store
.insert_event(&EventRow {
site: site.id.clone(),
ts: now,
kind: "click".into(),
visitor,
session: id,
pageview: String::new(),
path: js::head16(&pathname, 1000),
hostname: host.clone(),
title: String::new(),
name: link.slug.clone(),
props: None,
engaged_ms: 0,
scroll: None,
link: link.id.clone(),
})
.await?;
Ok::<(), Error>(())
}
.await;
if let Err(error) = recorded {
eprintln!("Runlight: could not record a link click {error}");
}
}
Ok(Some(Response::new(
Vec::new(),
302,
Headers::new()
.with("location", &link.url)
.with("cache-control", "no-store")
.with("referrer-policy", "no-referrer-when-downgrade"),
)))
}
async fn engagement(&self, site: &SiteRow, payload: &Payload, now: i64) -> Result<(), Error> {
if payload.engaged_ms <= 0 {
return Ok(());
}
let store = &self.0.store;
let Some(pageview) = store.pageview(&site.id, &payload.pageview_id).await? else { return Ok(()) };
if now - pageview.started_at >= EVENT_TAIL_MS {
return Ok(());
}
store.add_engagement(&pageview.session, payload.engaged_ms).await?;
if now - pageview.started_at > 3_600_000 {
store.touched_old_visit(&site.id, pageview.started_at, now - ROLLUP_DELAY_MS + 3_600_000).await?;
}
store
.insert_event(&EventRow {
site: site.id.clone(),
ts: now,
kind: "engagement".into(),
visitor: pageview.visitor,
session: pageview.session,
pageview: payload.pageview_id.clone(),
path: pageview.path,
hostname: pageview.hostname,
title: String::new(),
name: String::new(),
props: None,
engaged_ms: payload.engaged_ms,
scroll: payload.scroll,
link: String::new(),
})
.await?;
Ok(())
}
pub async fn observe(&self, request: &Request, at: Option<f64>) -> bool {
match self.observe_inner(request, at).await {
Ok(done) => done,
Err(error) => {
eprintln!("Runlight: could not record an AI agent fetch {error}");
false
}
}
}
async fn observe_inner(&self, request: &Request, at: Option<f64>) -> Result<bool, Error> {
if request.method != "GET" {
return Ok(false);
}
let Some(agent) = ai_agent(&request.headers.get("user-agent").unwrap_or_default()) else { return Ok(false) };
let Some(url) = Url::parse(&request.url) else { return Err(Error::Other("TypeError: Invalid URL".into())) };
let pathname = url.pathname();
if let Some(ext) = crate::re::group(js_re!(r"(?i)\.([a-z0-9]+)$"), &pathname, 1)
&& !["html", "htm", "md", "txt", "php"].contains(&ext.to_lowercase().as_str())
{
return Ok(false);
}
let host =
self.forwarded_host(request).or_else(|| request.headers.get("host")).unwrap_or_else(|| url.hostname());
self.init().await?;
let Some(site) = self.site_for(host.split(':').next().unwrap_or(&host), None) else { return Ok(false) };
let now = self.now();
if let Some(at) = at
&& at.is_finite()
&& at < (now - 7 * 86_400_000) as f64
{
return Ok(false);
}
let ts = match at {
Some(at) if at.is_finite() && at <= now as f64 => at.floor() as i64,
_ => now,
};
let mut props = Object::new();
props.set("company", agent.company);
props.set("kind", agent.kind);
self.0
.store
.insert_event(&EventRow {
site: site.id,
ts,
kind: "fetch".into(),
visitor: String::new(),
session: String::new(),
pageview: String::new(),
path: js::head16(&pathname, 1000),
hostname: strip_www(&url.hostname()),
title: String::new(),
name: agent.name.into(),
props: Some(props),
engaged_ms: 0,
scroll: None,
link: String::new(),
})
.await?;
Ok(true)
}
pub async fn check(&self) -> Result<Value, Error> {
let _turn = self.0.checking.lock().await;
self.init().await?;
let store = &self.0.store;
if self.0.managed_sites {
let sites = store.sites().await?;
self.state().configured = sites;
self.load_remotes().await?;
}
let overrides = store.site_overrides().await?;
{
let mut state = self.state();
state.overrides = overrides;
state.salts.clear();
}
let mut zones: Vec<String> = Vec::new();
for site in self.sites() {
if !zones.contains(&site.timezone) {
zones.push(site.timezone);
}
}
for zone in zones {
self.current_salts(self.now(), &zone).await?;
}
self.drop_old_salts(self.now()).await?;
self.idle().await;
{
let _turn = self.0.prune_turn.lock().await;
self.apply_retention(None).await?;
}
let due = {
let mut state = self.state();
let due = self.now() - state.optimized_at >= 86_400_000;
if due {
state.optimized_at = self.now();
}
due
};
if due {
store.optimize(false).await;
}
self.build_rollups().await?;
let reports = self.send_reports().await?;
Ok(crate::obj! { "ok" => true, "reports" => crate::obj! { "sent" => reports.0, "failed" => reports.1 } })
}
}
fn obj_zone(zone: &str, since: i64) -> String {
crate::obj! { "zone" => zone, "since" => since }.to_json()
}
fn site_among(sites: &[SiteRow], host: &str, id: Option<&str>) -> Option<SiteRow> {
if let Some(id) = id {
let site = sites.iter().find(|s| s.id == id)?;
return (site.hostnames.is_empty() || site.hostnames.iter().any(|h| h == host)).then(|| site.clone());
}
if sites.len() == 1 {
let only = &sites[0];
return (only.hostnames.is_empty() || only.hostnames.iter().any(|h| h == host)).then(|| only.clone());
}
sites.iter().find(|s| s.hostnames.iter().any(|h| h == host)).cloned()
}
pub fn is_domain(host: &str) -> bool {
let len = host.chars().count();
(1..=253).contains(&len) && test(js_re!(r"^([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$"), host)
}
fn minus_months(now: i64, months: i64) -> i64 {
let days = now.div_euclid(86_400_000);
let rest = now.rem_euclid(86_400_000);
let (y, m, d) = js::civil_from_days(days);
js::date_utc(y, m - 1 - months, d, 0, 0, 0, 0) + rest
}
fn not_found() -> Response {
Response::new("Not found", 404, Headers::new().with("content-type", "text/plain; charset=utf-8"))
}
fn busy(error: &Error) -> bool {
test(
js_re!(
r"(?i)timeout exceeded when trying to connect|connection timeout|no MySQL connection was free|SQLITE_BUSY|database is locked"
),
error.message(),
)
}