Fix batch requests to use batch endpoint instead of capturing events in serial, flatten properties

Cargo.toml

@@ -1,16 +1,7 @@
-[package]
-name = "posthog-rs"
-license = "MIT"
-version = "0.2.3"
-authors = ["christos <[email protected]>"]
-description = "An unofficial Rust client for Posthog (https://posthog.com/)."
-repository = "https://github.com/openquery-io/posthog-rs"
-edition = "2018"
+[workspace]
+members = [
+	"async",
+	"core",
+	"sync",
+]
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-reqwest = { version = "0.11.3", default-features = false, features = ["blocking", "rustls-tls"] }
-serde = { version = "1.0.125", features = ["derive"] }
-chrono = {version = "0.4.19", features = ["serde"] }
-serde_json = "1.0.64"

async/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "async-posthog"
+license = "MIT"
+version = "0.2.3"
+description = "An unofficial Rust client for Posthog (https://posthog.com/)."
+repository = "https://github.com/openquery-io/posthog-rs"
+edition = "2021"
+
+[dependencies]
+posthog-core = { path = "../core" }
+reqwest = { version = "0.11.3", default-features = false, features = ["json", "rustls-tls"] }
+tokio = { version = "1", features = ["full"] }
+serde = { version = "1.0.125", features = ["derive"] }
+serde_json = "1.0.64"
+thiserror = "1.0.38"

async/src/client.rs

@@ -0,0 +1,62 @@
+use posthog_core::event::{Event, InnerEvent, InnerEventBatch};
+use reqwest::{Client as HttpClient, Method};
+use serde::{de::DeserializeOwned, Serialize};
+
+use crate::client_options::ClientOptions;
+use crate::error::Error;
+
+pub struct Client {
+    options: ClientOptions,
+    http_client: HttpClient,
+}
+
+impl Client {
+    pub(crate) fn new(options: ClientOptions) -> Self {
+        let http_client = HttpClient::builder()
+            .timeout(options.timeout)
+            .build()
+            .unwrap(); // Unwrap here is as safe as `HttpClient::new`
+        Client {
+            options,
+            http_client,
+        }
+    }
+
+    async fn send_request<P: AsRef<str>, Body: Serialize, Res: DeserializeOwned>(
+        &self,
+        method: Method,
+        path: P,
+        body: &Body,
+    ) -> Result<Res, Error> {
+        let res = self
+            .http_client
+            .request(
+                method,
+                format!("{}{}", self.options.api_endpoint, path.as_ref()),
+            )
+            .json(body)
+            .send()
+            .await
+            .map_err(|source| Error::SendRequest { source })?
+            .error_for_status()
+            .map_err(|source| Error::ResponseStatus { source })?
+            .json::<Res>()
+            .await
+            .map_err(|source| Error::DecodeResponse { source })?;
+        Ok(res)
+    }
+
+    pub async fn capture(&self, event: Event) -> Result<(), Error> {
+        let inner_event = InnerEvent::new(event, self.options.api_key.clone());
+        self.send_request::<_, _, serde_json::Value>(Method::POST, "/capture/", &inner_event)
+            .await?;
+        Ok(())
+    }
+
+    pub async fn capture_batch(&self, events: Vec<Event>) -> Result<(), Error> {
+        let inner_event_batch = InnerEventBatch::new(events, self.options.api_key.clone());
+        self.send_request::<_, _, serde_json::Value>(Method::POST, "/batch/", &inner_event_batch)
+            .await?;
+        Ok(())
+    }
+}

async/src/client_options.rs

@@ -0,0 +1,42 @@
+use std::time::Duration;
+
+use crate::client::Client;
+
+const API_ENDPOINT: &str = "https://app.posthog.com";
+const TIMEOUT: Duration = Duration::from_millis(800); // This should be specified by the user
+
+pub struct ClientOptions {
+    pub(crate) api_endpoint: String,
+    pub(crate) api_key: String,
+    pub(crate) timeout: Duration,
+}
+
+impl ClientOptions {
+    pub fn new(api_key: impl ToString) -> ClientOptions {
+        ClientOptions {
+            api_endpoint: API_ENDPOINT.to_string(),
+            api_key: api_key.to_string(),
+            timeout: TIMEOUT,
+        }
+    }
+
+    pub fn api_endpoint(&mut self, api_endpoint: impl ToString) -> &mut Self {
+        self.api_endpoint = api_endpoint.to_string();
+        self
+    }
+
+    pub fn timeout(&mut self, timeout: Duration) -> &mut Self {
+        self.timeout = timeout;
+        self
+    }
+
+    pub fn build(self) -> Client {
+        Client::new(self)
+    }
+}
+
+impl From<&str> for ClientOptions {
+    fn from(api_key: &str) -> Self {
+        ClientOptions::new(api_key)
+    }
+}

async/src/error.rs

@@ -0,0 +1,11 @@
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("{source}")]
+    PostHogCore { source: posthog_core::error::Error },
+    #[error("send request: {source}")]
+    SendRequest { source: reqwest::Error },
+    #[error("response status: {source}")]
+    ResponseStatus { source: reqwest::Error },
+    #[error("decode response: {source}")]
+    DecodeResponse { source: reqwest::Error },
+}

async/src/lib.rs

@@ -0,0 +1,13 @@
+mod client;
+mod client_options;
+mod error;
+
+pub use client::Client;
+pub use client_options::ClientOptions;
+pub use error::Error;
+
+pub use posthog_core::event::{Event, Properties};
+
+pub fn client<C: Into<ClientOptions>>(options: C) -> Client {
+    options.into().build()
+}

async/tests/basic.rs

@@ -0,0 +1,42 @@
+use async_posthog::Event;
+use std::collections::HashMap;
+
+fn build_client() -> async_posthog::Client {
+    async_posthog::client(env!("POSTHOG_API_KEY"))
+}
+
+#[tokio::test]
+async fn capture() {
+    let client = build_client();
+
+    let mut child_map = HashMap::new();
+    child_map.insert("child_key1", "child_value1");
+
+    let mut event = Event::new("test async capture", "1234");
+    event.insert_prop("key1", "value1").unwrap();
+    event.insert_prop("key2", vec!["a", "b"]).unwrap();
+    event.insert_prop("key3", child_map).unwrap();
+
+    client.capture(event).await.unwrap();
+}
+
+#[tokio::test]
+async fn capture_batch() {
+    let client = build_client();
+
+    let events = (0..16)
+        .map(|_| {
+            let mut child_map = HashMap::new();
+            child_map.insert("child_key1", "child_value1");
+
+            let mut event = Event::new("test async capture batch", "1234");
+            event.insert_prop("key1", "value1").unwrap();
+            event.insert_prop("key2", vec!["a", "b"]).unwrap();
+            event.insert_prop("key3", child_map).unwrap();
+
+            event
+        })
+        .collect::<Vec<_>>();
+
+    client.capture_batch(events).await.unwrap();
+}

core/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "posthog-core"
+license = "MIT"
+version = "0.1.0"
+description = "An unofficial Rust client for Posthog (https://posthog.com/)."
+repository = "https://github.com/openquery-io/posthog-rs"
+edition = "2021"
+
+[dependencies]
+chrono = {version = "0.4.19", features = ["serde"] }
+serde = { version = "1.0.125", features = ["derive"] }
+serde_json = "1.0.64"
+thiserror = "1.0.38"

core/src/error.rs

@@ -0,0 +1,8 @@
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("serialization: {source}")]
+    Serialization {
+        #[from]
+        source: serde_json::Error,
+    },
+}

core/src/event.rs

@@ -0,0 +1,75 @@
+use chrono::NaiveDateTime;
+use serde::Serialize;
+use std::collections::HashMap;
+
+use crate::error::Error;
+
+#[derive(Serialize)]
+pub struct InnerEvent {
+    api_key: String,
+    #[serde(flatten)]
+    event: Event,
+}
+
+impl InnerEvent {
+    pub fn new(event: Event, api_key: String) -> Self {
+        Self { api_key, event }
+    }
+}
+
+#[derive(Serialize)]
+pub struct InnerEventBatch {
+    api_key: String,
+    batch: Vec<Event>,
+}
+
+impl InnerEventBatch {
+    pub fn new(batch: Vec<Event>, api_key: String) -> Self {
+        Self { api_key, batch }
+    }
+}
+
+#[derive(Serialize, Debug, PartialEq, Eq)]
+pub struct Event {
+    event: String,
+    properties: Properties,
+    timestamp: Option<NaiveDateTime>,
+}
+
+#[derive(Serialize, Debug, PartialEq, Eq)]
+pub struct Properties {
+    distinct_id: String,
+    #[serde(flatten)]
+    props: HashMap<String, serde_json::Value>,
+}
+
+impl Properties {
+    fn new<S: Into<String>>(distinct_id: S) -> Self {
+        Self {
+            distinct_id: distinct_id.into(),
+            props: Default::default(),
+        }
+    }
+}
+
+impl Event {
+    pub fn new<S: Into<String>>(event: S, distinct_id: S) -> Self {
+        Self {
+            event: event.into(),
+            properties: Properties::new(distinct_id),
+            timestamp: None,
+        }
+    }
+
+    /// Errors if `prop` fails to serialize
+    pub fn insert_prop<K: Into<String>, P: Serialize>(
+        &mut self,
+        key: K,
+        prop: P,
+    ) -> Result<(), Error> {
+        let as_json =
+            serde_json::to_value(prop).map_err(|source| Error::Serialization { source })?;
+        let _ = self.properties.props.insert(key.into(), as_json);
+        Ok(())
+    }
+}

core/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod error;
+pub mod event;