feat: add all edgee event, and plug enterprise data-collection-api

Cargo.toml

@@ -14,6 +14,7 @@ homepage = "https://www.edgee.cloud"
 addr = "0.15.6"
 aes = "0.8.4"
 anyhow = "1.0.83"
+base64 = "0.22.1"
 brotli = "6.0.0"
 bytes = "1.6.0"
 cbc = { version = "0.1.2", features = ["std"] }

README.md

@@ -170,7 +170,7 @@ You simply need to add a new session to your configuration pointing to the WebAs
 
 ```toml
 # edgee.toml
-[[destinations.data_collection]]
+[[components.data_collection]]
 name = "amplitude"
 component = "/var/edgee/wasm/amplitude.wasm"
 credentials.amplitude_api_key = "YOUR-API-KEY"

edgee.sample.toml

@@ -13,17 +13,17 @@ key = "local/cert/edgee.key"
 [monitor]
 address = "0.0.0.0:8222"
 
-[[destinations.data_collection]]
+[[components.data_collection]]
 name = "amplitude"
 component = "local/wasm/amplitude.wasm"
 credentials.amplitude_api_key = "..."
 
-[[destinations.data_collection]]
+[[components.data_collection]]
 name = "google analytics"
 component = "local/wasm/ga.wasm"
 credentials.ga_measurement_id = "..."
 
-[[destinations.data_collection]]
+[[components.data_collection]]
 name = "segment"
 component = "local/wasm/segment.wasm"
 credentials.segment_project_id = "..."

src/config/config.rs

@@ -1,6 +1,6 @@
 use std::collections::{HashMap, HashSet};
 use std::path::Path;
-use serde::Deserialize;
+use serde::{Deserialize};
 use tokio::sync::OnceCell;
 
 static CONFIG: OnceCell<StaticConfiguration> = OnceCell::const_new();
@@ -26,7 +26,7 @@ pub struct StaticConfiguration {
     pub routing: Vec<RoutingConfiguration>,
 
     #[serde(default)]
-    pub destinations: DestinationConfiguration,
+    pub components: ComponentsConfiguration,
 }
 
 fn default_log_config() -> LogConfiguration {
@@ -41,6 +41,8 @@ fn default_compute_config() -> ComputeConfiguration {
         max_decompressed_body_size: default_max_decompressed_body_size(),
         max_compressed_body_size: default_max_compressed_body_size(),
         proxy_only: false,
+        data_collection_api_key: None,
+        data_collection_api_url: None,
     }
 }
 
@@ -147,7 +149,7 @@ pub struct BackendConfiguration {
 }
 
 #[derive(Deserialize, Debug, Default, Clone)]
-pub struct DestinationConfiguration {
+pub struct ComponentsConfiguration {
     pub data_collection: Vec<DataCollectionConfiguration>,
 }
 
@@ -174,6 +176,8 @@ pub struct ComputeConfiguration {
     pub max_compressed_body_size: usize,
     #[serde(default)]
     pub proxy_only: bool,
+    pub data_collection_api_key: Option<String>,
+    pub data_collection_api_url: Option<String>,
 }
 
 fn default_cookie_name() -> String {
@@ -239,7 +243,7 @@ pub fn init_test_config() {
         monitor: Some(MonitorConfiguration { address: "127.0.0.1:9090".to_string() }),
         routing: vec![],
         compute: default_compute_config(),
-        destinations: DestinationConfiguration::default(),
+        components: ComponentsConfiguration::default(),
     };
 
     if CONFIG.get().is_none() {

src/proxy/compute/compute.rs

@@ -1,25 +1,22 @@
-use std::collections::HashMap;
+use bytes::Bytes;
 use crate::config::config;
 use crate::proxy::compute::html::{parse_html, Document};
 use crate::proxy::proxy::set_edgee_header;
 use crate::tools::edgee_cookie;
-use crate::proxy::compute::data_collection::{data_collection, components};
+use crate::proxy::compute::data_collection::data_collection;
 use http::response::Parts;
 use http::uri::PathAndQuery;
 use http::HeaderMap;
-use log::{debug, warn};
-use std::net::SocketAddr;
-use http::header::{ACCEPT_LANGUAGE, REFERER, USER_AGENT};
-use crate::proxy::compute::data_collection::data_collection::{Payload, Session};
-use crate::tools::real_ip::Realip;
+use log::{warn};
+use crate::tools::edgee_cookie::EdgeeCookie;
 
 pub async fn html_handler(
     body: &String,
     host: &String,
     path: &PathAndQuery,
     request_headers: &HeaderMap,
     proto: &str,
-    remote_addr: &SocketAddr,
+    client_ip: &String,
     response_parts: &mut Parts,
     response_headers: &HeaderMap,
 ) -> Result<Document, &'static str>
@@ -42,12 +39,10 @@ pub async fn html_handler(
             if cookie.is_none() {
                 set_edgee_header(response_parts, "compute-aborted(no-cookie)");
             } else {
-                let payload = data_collection::process_from_html(&document, &cookie.unwrap(), proto, &host, &path, &response_headers, remote_addr);
-                let uuid = payload.uuid.clone();
-                if let Err(err) = components::send_data_collection(payload).await {
-                    tracing::warn!(?err, "failed to send data collection payload");
+                let data_collection_trace_uuid = data_collection::process_from_html(&document, &cookie.unwrap(), proto, &host, &path, &request_headers, client_ip).await;
+                if data_collection_trace_uuid.is_some() {
+                    document.trace_uuid = data_collection_trace_uuid.unwrap();
                 }
-                document.trace_uuid = uuid;
             }
         }
         Err(reason) => {
@@ -58,147 +53,8 @@ pub async fn html_handler(
     Ok(document)
 }
 
-pub async fn json_handler(payload: &mut Payload, host: &String, path: &PathAndQuery, request_headers: &HeaderMap, remote_addr: &SocketAddr, response_headers: &mut HeaderMap) {
-    let cookie = edgee_cookie::get_or_set(&request_headers, response_headers, &host);
-
-    payload.uuid = uuid::Uuid::new_v4().to_string();
-    payload.timestamp = chrono::Utc::now();
-
-    let user_id = cookie.id.to_string();
-    payload.identify.edgee_id = user_id.clone();
-    payload.session = Session {
-        session_id: cookie.ss.timestamp().to_string(),
-        previous_session_id: cookie
-            .ps
-            .map(|t| t.timestamp().to_string())
-            .unwrap_or_default(),
-        session_count: cookie.sc,
-        session_start: cookie.ss == cookie.ls,
-        first_seen: cookie.fs,
-        last_seen: cookie.ls,
-    };
-
-    if payload.page.referrer.is_empty() {
-        let referrer = request_headers
-            .get(REFERER)
-            .and_then(|h| h.to_str().ok())
-            .map(String::from)
-            .unwrap_or_default();
-        payload.page.referrer = referrer;
-    }
-
-    payload.client.user_agent = request_headers
-        .get(USER_AGENT)
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.x_forwarded_for = request_headers
-        .get("x-forwarded-for")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.user_agent_architecture = request_headers
-        .get("sec-ch-ua-arch")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.user_agent_bitness = request_headers
-        .get("sec-ch-ua-bitness")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.user_agent_full_version_list = request_headers
-        .get("sec-ch-ua")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.user_agent_mobile = request_headers
-        .get("sec-ch-ua-mobile")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.user_agent_model = request_headers
-        .get("sec-ch-ua-model")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.os_name = request_headers
-        .get("sec-ch-ua-platform")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.client.os_version = request_headers
-        .get("sec-ch-ua-platform-version")
-        .and_then(|h| h.to_str().ok())
-        .map(String::from)
-        .unwrap_or_default();
-
-    // client ip
-    let realip = Realip::new();
-    payload.client.ip = realip.get_from_request(&remote_addr, &request_headers);
-
-    payload.client.locale = preferred_language(&request_headers);
-
-    let map: HashMap<String, String> =
-        url::form_urlencoded::parse(path.query().unwrap_or("").as_bytes())
-            .into_owned()
-            .collect();
-
-    payload.campaign.name = map
-        .get("utm_campaign")
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.campaign.source =
-        map.get("utm_source").map(String::from).unwrap_or_default();
-
-    payload.campaign.medium =
-        map.get("utm_medium").map(String::from).unwrap_or_default();
-
-    payload.campaign.term = map.get("utm_term").map(String::from).unwrap_or_default();
-
-    payload.campaign.content =
-        map.get("utm_content").map(String::from).unwrap_or_default();
-
-    payload.campaign.creative_format = map
-        .get("utm_creative_format")
-        .map(String::from)
-        .unwrap_or_default();
-
-    payload.campaign.marketing_tactic = map
-        .get("utm_marketing_tactic")
-        .map(String::from)
-        .unwrap_or_default();
-
-    debug!("data collection payload: {:?}", payload);
-    if let Err(err) = components::send_data_collection(payload.clone()).await {
-        warn!("{} {}", err, "failed to process data collection");
-    }
-}
-
-
-fn preferred_language(headers: &HeaderMap) -> String {
-    let accept_language_header = headers
-        .get(ACCEPT_LANGUAGE)
-        .and_then(|h| h.to_str().ok())
-        .unwrap_or_default();
-    let languages = accept_language_header.split(",");
-    let lang = "en-us".to_string();
-    for l in languages {
-        let lang = l.split(";").next().unwrap_or("").trim();
-        if !lang.is_empty() {
-            return lang.to_lowercase();
-        }
-    }
-    lang
+pub async fn json_handler(body: &Bytes, cookie: &EdgeeCookie, path: &PathAndQuery, request_headers: &HeaderMap, client_ip: &String) {
+    data_collection::process_from_json(body, &cookie, &path, &request_headers, client_ip).await;
 }
 
 /// Processes the payload of a request under certain conditions.