Google Analytics 4

use ga4.qweruse ga4_check, ga4_event from ga4      // Quillo 0.6: only these

ureq = "2"jsonwebtoken = "9"serde_json = { version = "1", features = ["preserve_order"] }

// Google Analytics 4 from the server: events checked and sent, the validation endpoint, and reports// read back as rows.////   GA4_MEASUREMENT_ID=G-... GA4_API_SECRET=... GA4_PROPERTY_ID=123456789 \//   GOOGLE_APPLICATION_CREDENTIALS=key.json quillo run ga4.qwe//   then open /send, /checks, /validate, /report, /realtime and /failures//// The API secret is made under Data Streams in GA4's admin. Add the service account's address to the// property's access. GA4_MP_URL and GA4_DATA_API_URL reach the EU, or a test server.use "ga4.qwer"route GET /send = {    one: ga4_send("555.1234567890", [ga4_event("purchase", { currency: "EUR", value: 12.5, transaction_id: "T1" })], { user_id: "user-1", user_properties: { plan: "pro" } })?,    many: ga4_send("555.1234567890", map(range(0, 60), (i) => ga4_event("page_view", { n: i })), {})?,    not_sent: ga4_send("555.1234567890", [ga4_event("first_open", {})], {}).error,    no_client: ga4_send("", [ga4_event("x", {})], {}).error,    none: ga4_send("555.1", [], {}).error,}route GET /checks = {    fit: ga4_check("c", [ga4_event("sign_up", { method: "email" })], {}),    reserved: ga4_check("c", [ga4_event("session_start", {})], {}),    long: ga4_check("c", [ga4_event(repeat("a", 41), {})], {}),    digit: ga4_check("c", [ga4_event("1st", {})], {}),    prefix: ga4_check("c", [ga4_event("google_thing", { ga_x: 1 })], {}),    value: ga4_check("c", [ga4_event("x", { note: repeat("n", 101) })], {}),    map_param: ga4_check("c", [ga4_event("x", { nested: { a: 1 } })], {}),    user: ga4_check("c", [ga4_event("x", {})], { user_properties: { user_id: "1", favourite_colour_of_my_cat: "x", plan: repeat("p", 37) } }),    old: ga4_check("c", [ga4_event("x", {})], { timestamp_micros: 1600000000000000 }),    params: len(ga4_check("c", [ga4_event("x", reduce(range(0, 26), {}, (m, i) => set(m, `p${i}`, i)))], {})),}route GET /validate = {    fit: ga4_validate("555.1", [ga4_event("sign_up", {})], {})?,    wrong: ga4_validate("555.1", [ga4_event("_bad", {})], {})?,}route GET /report {    let r = ga4_report("", { dimensions: ["country"], metrics: ["activeUsers", "bounceRate", "totalRevenue"], start: "7daysAgo", end: "yesterday", orderBys: [{ metric: { metricName: "activeUsers" }, desc: true }] }, 0)?    let paged = ga4_report("123456789", { dimensions: ["pagePath"], metrics: ["screenPageViews"], page_size: 500 }, 0)?    {        first: r.rows[0], count: r.row_count, quota: r.quota, currency: r.currency, zone: r.time_zone,        paged: len(paged.rows), capped: len(ga4_report("properties/123456789", { dimensions: ["pagePath"], metrics: ["screenPageViews"], page_size: 500 }, 700)?.rows),        empty: ga4_report("", { metrics: ["sessions"], dimensions: ["emptySet"] }, 0)?,        no_metrics: ga4_report("", {}, 0).error,        g_id: ga4_report("G-ABC123", { metrics: ["sessions"] }, 0).error,    }}route GET /realtime = {    now: ga4_realtime("", { dimensions: ["unifiedScreenName"], metrics: ["activeUsers"] })?,}route GET /failures = {    bad_metric: ga4_report("", { metrics: ["nonsense"] }, 0).error,    no_access: ga4_report("999", { metrics: ["sessions"] }, 0).error,    spent: ga4_report("888", { metrics: ["sessions"] }, 0).error,    waited: ga4_report("777", { metrics: ["sessions"] }, 0)?.row_count,}

//! Google Analytics 4 from the server: the Measurement Protocol with its limits and reserved names checked first, since the live endpoint accepts anything and answers 204, its validation endpoint read back, and Data API reports by service-account JWT, returned as rows of maps with typed numbers, paged by offset, with the property's quota.//! version: 1.0.0//!//! [dependencies]//! ureq = "2"//! jsonwebtoken = "9"//! serde_json = { version = "1", features = ["preserve_order"] }use serde_json::{json, Map, Value};use std::sync::Mutex;use std::time::Duration;// GA4_MEASUREMENT_ID (G-...) and GA4_API_SECRET send events; GA4_MP_URL is the Measurement Protocol// (https://www.google-analytics.com; https://region1.google-analytics.com keeps data in the EU).// GA4_PROPERTY_ID is the numeric property for reports, GA4_DATA_API_URL the Data API// (https://analyticsdata.googleapis.com). GOOGLE_SERVICE_ACCOUNT_JSON holds the service account's// JSON, or GOOGLE_APPLICATION_CREDENTIALS its file; add its address to the property's access.const SCOPE: &str = "https://www.googleapis.com/auth/analytics.readonly";const RESERVED: &[&str] = &["ad_activeview", "ad_click", "ad_exposure", "ad_query", "ad_reward", "adunit_exposure", "app_clear_data", "app_exception", "app_install", "app_remove", "app_store_refund", "app_update", "app_upgrade", "dynamic_link_app_open", "dynamic_link_app_update", "dynamic_link_first_open", "error", "firebase_campaign", "firebase_in_app_message_action", "firebase_in_app_message_dismiss", "firebase_in_app_message_impression", "first_open", "first_visit", "in_app_purchase", "notification_dismiss", "notification_foreground", "notification_open", "notification_receive", "notification_send", "os_update", "session_start", "user_engagement"];const RESERVED_USER: &[&str] = &["first_open_time", "first_visit_time", "last_deep_link_referrer", "user_id", "first_open_after_install"];fn setting(k: &str) -> Option<String> {    std::env::var(k).ok().filter(|v| !v.is_empty())}fn now() -> i64 {    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_secs() as i64)}fn enc(s: &str) -> String {    s.bytes().map(|b| if b.is_ascii_alphanumeric() || b"-_.~".contains(&b) { (b as char).to_string() } else { format!("%{:02X}", b) }).collect()}fn name_problem(name: &str, limit: usize) -> Option<String> {    if name.is_empty() {        return Some("is empty".into());    }    if name.chars().count() > limit {        return Some(format!("is over {} characters", limit));    }    if !name.starts_with(|c: char| c.is_ascii_alphabetic()) {        return Some("must start with a letter".into());    }    if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {        return Some("may hold only letters, digits and _".into());    }    for p in ["firebase_", "ga_", "google_", "gtag."] {        if name.starts_with(p) {            return Some(format!("starts with the reserved `{}`", p));        }    }    None}/// Everything wrong with these events and options by GA4's own rules: at most 25 events, event names of 40 characters that are not reserved, 25 parameters of 40-character names and 100-character values, 25 user properties of 24-character names and 36-character values, and a timestamp at most 72 hours old. Gives a list of problems, empty when fit.pub fn ga4_check(client_id: &str, events: Vec<serde_json::Value>, options: serde_json::Value) -> Vec<String> {    let mut out = vec![];    if client_id.is_empty() {        out.push("client_id is required".to_string());    }    if events.is_empty() {        out.push("at least one event is required".into());    }    for (i, e) in events.iter().enumerate() {        let name = e["name"].as_str().unwrap_or("");        if let Some(p) = name_problem(name, 40) {            out.push(format!("event {}: the name `{}` {}", i, name, p));        } else if RESERVED.contains(&name) {            out.push(format!("event {}: `{}` is a reserved name", i, name));        }        let params = e["params"].as_object().cloned().unwrap_or_default();        if params.len() > 25 {            out.push(format!("event {}: {} parameters, at most 25", i, params.len()));        }        for (k, v) in &params {            if let Some(p) = name_problem(k, 40) {                out.push(format!("event {}: the parameter `{}` {}", i, k, p));            }            if k == "firebase_conversion" {                out.push(format!("event {}: `firebase_conversion` is reserved", i));            }            if v.as_str().map_or(false, |s| s.chars().count() > 100) {                out.push(format!("event {}: parameter `{}` is over 100 characters", i, k));            }            if v.is_object() {                out.push(format!("event {}: parameter `{}` is a map; only `items` holds a list", i, k));            }        }    }    let up = options["user_properties"].as_object().cloned().unwrap_or_default();    if up.len() > 25 {        out.push(format!("{} user properties, at most 25", up.len()));    }    for (k, v) in &up {        if let Some(p) = name_problem(k, 24) {            out.push(format!("the user property `{}` {}", k, p));        } else if RESERVED_USER.contains(&k.as_str()) {            out.push(format!("`{}` is a reserved user property", k));        }        if v.as_str().map_or(false, |s| s.chars().count() > 36) {            out.push(format!("user property `{}` is over 36 characters", k));        }    }    if let Some(t) = options["timestamp_micros"].as_i64() {        if now() - t / 1_000_000 > 72 * 3600 {            out.push("timestamp_micros is over 72 hours old: GA4 drops it".into());        }        if t / 1_000_000 > now() + 900 {            out.push("timestamp_micros is in the future".into());        }    }    out}fn payload(client_id: &str, events: &[Value], options: &Value) -> Value {    let mut b = json!({ "client_id": client_id, "events": events });    if let Some(u) = options["user_id"].as_str() {        b["user_id"] = json!(u);    }    if let Some(t) = options["timestamp_micros"].as_i64() {        b["timestamp_micros"] = json!(t);    }    if let Some(up) = options["user_properties"].as_object() {        let mut m = Map::new();        for (k, v) in up {            m.insert(k.clone(), json!({ "value": v }));        }        b["user_properties"] = Value::Object(m);    }    for k in ["non_personalized_ads", "consent"] {        if let Some(v) = options.get(k) {            b[k] = v.clone();        }    }    b}fn mp(path: &str, body: &Value) -> Result<(u16, String), String> {    let id = setting("GA4_MEASUREMENT_ID").ok_or("GA4_MEASUREMENT_ID is not set")?;    let secret = setting("GA4_API_SECRET").ok_or("GA4_API_SECRET is not set")?;    let host = setting("GA4_MP_URL").unwrap_or_else(|| "https://www.google-analytics.com".into());    let url = format!("{}{}?measurement_id={}&api_secret={}", host.trim_end_matches('/'), path, enc(&id), enc(&secret));    let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(20)).build();    let text = body.to_string();    for attempt in 0..4u32 {        match agent.post(&url).set("Content-Type", "application/json").send_string(&text) {            Ok(r) => {                let c = r.status();                return Ok((c, r.into_string().unwrap_or_default()));            }            Err(ureq::Error::Status(429 | 500 | 502 | 503, _)) if attempt < 3 => std::thread::sleep(Duration::from_millis(300u64 << attempt)),            Err(ureq::Error::Status(c, r)) => return Err(format!("GA4 answered {}: {}", c, r.into_string().unwrap_or_default().chars().take(300).collect::<String>())),            Err(e) => return Err(format!("cannot reach GA4: {}", e)),        }    }    Err("GA4 stayed busy through four tries".into())}/// One event, `ga4_event("purchase", { value: 12.5, currency: "EUR" })`, as the Measurement Protocol wants it.pub fn ga4_event(name: &str, params: serde_json::Value) -> serde_json::Value {    json!({ "name": name, "params": if params.is_object() { params } else { json!({}) } })}/// Sends events for `client_id` (the browser's _ga id, or one of yours): `options` may hold `user_id`, `timestamp_micros`, `user_properties` (a plain map), `consent`, `non_personalized_ads`. The live endpoint answers 204 to anything, even a bad secret, so ga4_check runs first and any problem is an error. More than 25 are sent in groups. Gives { sent, requests }.pub fn ga4_send(client_id: &str, events: Vec<serde_json::Value>, options: serde_json::Value) -> Result<serde_json::Value, String> {    let mut problems = vec![];    for chunk in events.chunks(25) {        problems.extend(ga4_check(client_id, chunk.to_vec(), options.clone()));    }    if events.is_empty() {        problems.push("at least one event is required".into());    }    if !problems.is_empty() {        return Err(problems.join("; "));    }    let mut requests = 0;    for chunk in events.chunks(25) {        mp("/mp/collect", &payload(client_id, chunk, &options))?;        requests += 1;    }    Ok(json!({ "sent": events.len(), "requests": requests }))}/// Asks GA4's validation endpoint about a payload, without recording it: gives its messages, each { fieldPath, description, validationCode }, empty when fit. Use it while writing an integration, not on live traffic.pub fn ga4_validate(client_id: &str, events: Vec<serde_json::Value>, options: serde_json::Value) -> Result<Vec<serde_json::Value>, String> {    let (_, t) = mp("/debug/mp/collect", &payload(client_id, &events, &options))?;    let v: Value = serde_json::from_str(&t).map_err(|_| format!("GA4's validation answer is not JSON: {}", t.chars().take(200).collect::<String>()))?;    Ok(v["validationMessages"].as_array().cloned().unwrap_or_default())}fn credentials() -> Result<Value, String> {    let text = match (setting("GOOGLE_SERVICE_ACCOUNT_JSON"), setting("GOOGLE_APPLICATION_CREDENTIALS")) {        (Some(t), _) => t,        (None, Some(p)) => std::fs::read_to_string(&p).map_err(|e| format!("{}: {}", p, e))?,        _ => return Err("set GOOGLE_SERVICE_ACCOUNT_JSON or GOOGLE_APPLICATION_CREDENTIALS".into()),    };    let v: Value = serde_json::from_str(&text).map_err(|e| format!("the service account is not JSON: {}", e))?;    for k in ["client_email", "private_key"] {        if !v[k].is_string() {            return Err(format!("the service account has no {}", k));        }    }    Ok(v)}static TOKEN: Mutex<Option<(String, i64)>> = Mutex::new(None);fn token() -> Result<String, String> {    if let Some((t, until)) = TOKEN.lock().unwrap_or_else(|e| e.into_inner()).as_ref() {        if *until > now() + 300 {            return Ok(t.clone());        }    }    let sa = credentials()?;    let uri = sa["token_uri"].as_str().unwrap_or("https://oauth2.googleapis.com/token").to_string();    let key = jsonwebtoken::EncodingKey::from_rsa_pem(sa["private_key"].as_str().unwrap_or("").as_bytes()).map_err(|e| format!("the service account's private key: {}", e))?;    let mut head = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::RS256);    head.kid = sa["private_key_id"].as_str().map(str::to_string);    let t = now();    let jwt = jsonwebtoken::encode(&head, &json!({ "iss": sa["client_email"], "scope": SCOPE, "aud": uri, "iat": t, "exp": t + 3600 }), &key).map_err(|e| e.to_string())?;    let res = ureq::post(&uri).timeout(Duration::from_secs(20)).send_form(&[("grant_type", "urn:ietf:params:oauth:grant-type:jwt-bearer"), ("assertion", &jwt)]);    let text = match res {        Ok(r) => r.into_string().map_err(|e| e.to_string())?,        Err(ureq::Error::Status(c, r)) => {            let b = r.into_string().unwrap_or_default();            let v: Value = serde_json::from_str(&b).unwrap_or(Value::Null);            return Err(format!("the token server answered {}: {} {}", c, v["error"].as_str().unwrap_or(""), v["error_description"].as_str().unwrap_or(&b)));        }        Err(e) => return Err(format!("cannot reach the token server: {}", e)),    };    let v: Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;    let tok = v["access_token"].as_str().ok_or("the token server gave no access_token")?.to_string();    *TOKEN.lock().unwrap_or_else(|e| e.into_inner()) = Some((tok.clone(), t + v["expires_in"].as_i64().unwrap_or(3600)));    Ok(tok)}fn data_call(method_name: &str, property: &str, body: &Value) -> Result<Value, String> {    let id = if property.is_empty() { setting("GA4_PROPERTY_ID").ok_or("no property: pass one, or set GA4_PROPERTY_ID")? } else { property.trim_start_matches("properties/").to_string() };    if id.is_empty() || !id.bytes().all(|b| b.is_ascii_digit()) {        return Err(format!("`{}` is not a property: the number in GA4's admin, not the G- id", id));    }    let host = setting("GA4_DATA_API_URL").unwrap_or_else(|| "https://analyticsdata.googleapis.com".into());    let url = format!("{}/v1beta/properties/{}:{}", host.trim_end_matches('/'), id, method_name);    let tok = token()?;    let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(90)).build();    for attempt in 0..4u32 {        match agent.post(&url).set("Authorization", &format!("Bearer {}", tok)).set("Content-Type", "application/json").send_string(&body.to_string()) {            Ok(r) => return serde_json::from_str(&r.into_string().map_err(|e| e.to_string())?).map_err(|e| e.to_string()),            Err(ureq::Error::Status(c, r)) => {                let t = r.into_string().unwrap_or_default();                let v: Value = serde_json::from_str(&t).unwrap_or(Value::Null);                let m = v["error"]["message"].as_str().unwrap_or(&t).to_string();                let status = v["error"]["status"].as_str().unwrap_or("");                // Spent tokens come back on the hour or the day: a retry cannot help those.                let tokens = m.to_lowercase().contains("tokens");                if (c == 503 || (c == 429 && !tokens)) && attempt < 3 {                    std::thread::sleep(Duration::from_millis(500u64 << attempt));                    continue;                }                let hint = if c == 403 { " (add the service account's address to the property's access)" } else { "" };                return Err(format!("GA4 {} {}: {}{}", c, status, m, hint));            }            Err(e) => return Err(format!("cannot reach the Data API: {}", e)),        }    }    Err("the Data API stayed busy through four tries".into())}fn names(v: &Value) -> Value {    Value::Array(v.as_array().map(|l| l.iter().map(|x| if x.is_string() { json!({ "name": x }) } else { x.clone() }).collect()).unwrap_or_default())}fn typed(text: &str, kind: &str) -> Value {    match kind {        "TYPE_INTEGER" => text.parse::<i64>().map_or(json!(text), |n| json!(n)),        "" | "METRIC_TYPE_UNSPECIFIED" => json!(text),        _ => text.parse::<f64>().map_or(json!(text), |n| json!(n)),    }}fn rows_of(v: &Value) -> Vec<Value> {    let dims: Vec<String> = v["dimensionHeaders"].as_array().into_iter().flatten().filter_map(|h| h["name"].as_str().map(str::to_string)).collect();    let mets: Vec<(String, String)> = v["metricHeaders"].as_array().into_iter().flatten().map(|h| (h["name"].as_str().unwrap_or("").to_string(), h["type"].as_str().unwrap_or("").to_string())).collect();    v["rows"].as_array().into_iter().flatten().map(|r| {        let mut m = Map::new();        for (i, d) in dims.iter().enumerate() {            m.insert(d.clone(), r["dimensionValues"][i]["value"].clone());        }        for (i, (n, k)) in mets.iter().enumerate() {            m.insert(n.clone(), typed(r["metricValues"][i]["value"].as_str().unwrap_or(""), k));        }        Value::Object(m)    }).collect()}/// A report: `request` has `dimensions` and `metrics` (names, or the API's maps), `start` and `end` (dates, or "7daysAgo", "yesterday", "today"), and any other field of runReport (`dimensionFilter`, `orderBys`...). `max_rows` caps it (0: all), and `page_size` in `request` sets a page. Gives { rows, row_count, quota, currency, time_zone }, `rows` as maps with metrics as numbers.pub fn ga4_report(property: &str, request: serde_json::Value, max_rows: i64) -> Result<serde_json::Value, String> {    let mut b = if request.is_object() { request } else { return Err("a report request must be a map".into()) };    if !b["metrics"].is_array() {        return Err("a report needs a list of metrics".into());    }    let start = b.as_object_mut().and_then(|m| m.remove("start")).unwrap_or(json!("28daysAgo"));    let end = b.as_object_mut().and_then(|m| m.remove("end")).unwrap_or(json!("yesterday"));    let page_size = b.as_object_mut().and_then(|m| m.remove("page_size")).and_then(|v| v.as_i64()).unwrap_or(100_000).clamp(1, 250_000);    if b.get("dateRanges").is_none() {        b["dateRanges"] = json!([{ "startDate": start, "endDate": end }]);    }    b["metrics"] = names(&b["metrics"]);    if b.get("dimensions").is_some() {        b["dimensions"] = names(&b["dimensions"]);    }    b["returnPropertyQuota"] = json!(true);    let mut rows = vec![];    let mut offset = 0;    loop {        b["limit"] = json!(page_size);        b["offset"] = json!(offset);        let v = data_call("runReport", property, &b)?;        let page = rows_of(&v);        let n = page.len() as i64;        rows.extend(page);        offset += n;        let total = v["rowCount"].as_i64().unwrap_or(0);        let full = n == 0 || offset >= total || (max_rows > 0 && offset >= max_rows);        if full {            if max_rows > 0 {                rows.truncate(max_rows as usize);            }            let q = &v["propertyQuota"];            return Ok(json!({ "rows": rows, "row_count": total, "quota": { "tokens_per_day": q["tokensPerDay"]["remaining"], "tokens_per_hour": q["tokensPerHour"]["remaining"] }, "currency": v["metadata"]["currencyCode"], "time_zone": v["metadata"]["timeZone"] }));        }    }}/// The last 30 minutes: `request` as for ga4_report, without dates. Gives { rows, row_count }.pub fn ga4_realtime(property: &str, request: serde_json::Value) -> Result<serde_json::Value, String> {    let mut b = if request.is_object() { request } else { return Err("a report request must be a map".into()) };    b["metrics"] = names(&b["metrics"]);    if b.get("dimensions").is_some() {        b["dimensions"] = names(&b["dimensions"]);    }    let v = data_call("runRealtimeReport", property, &b)?;    Ok(json!({ "rows": rows_of(&v), "row_count": v["rowCount"] }))}