reflow_api_services 0.2.0

#![allow(clippy::all, unused_imports, dead_code)]

//! Auto-generated API actors for Amplitude
//!
//! Service: Amplitude (amplitude)
//! Product analytics and user behavior platform
//!
//! Required env var: AMPLITUDE_API_KEY (API key)
//!
//! Generated by api-schema-gen codegen — do not edit manually.

use crate::{Actor, ActorBehavior, Message, Port};
use anyhow::{Error, Result};
use reflow_actor::{message::EncodableValue, ActorContext};
use reflow_actor_macro::actor;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::time::Duration;

const BASE_URL: &str = "https://api2.amplitude.com/2";
const ENV_KEY: &str = "AMPLITUDE_API_KEY";

/// Apply authentication to the request builder.
fn apply_auth(
    config: &reflow_actor::ActorConfig,
    mut builder: reqwest::RequestBuilder,
) -> Result<reqwest::RequestBuilder> {
    let credential = config
        .get_config_or_env(ENV_KEY)
        .ok_or_else(|| anyhow::anyhow!("Missing env var: {}", ENV_KEY))?;
    builder = builder.header("Authorization", format!("Bearer {}", credential));
    Ok(builder)
}

/// send event via Amplitude API
///
/// Method: POST /httpapi
#[actor(
    AmplitudeSendEventActor,
    inports::<100>(api_key, events),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_send_event(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/httpapi".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.post(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut body = serde_json::Map::new();
    if let Some(val) = inputs.get("api_key") {
        body.insert("api_key".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("events") {
        body.insert("events".to_string(), val.clone().into());
    }
    if !body.is_empty() {
        builder = builder.json(&serde_json::Value::Object(body));
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("POST /httpapi failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Batch upload events (up to 2000 events per request)
///
/// Method: POST /batch
#[actor(
    AmplitudeCreateBatchEventActor,
    inports::<100>(api_key, events, options),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_create_batch_event(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/batch".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.post(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut body = serde_json::Map::new();
    if let Some(val) = inputs.get("api_key") {
        body.insert("api_key".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("events") {
        body.insert("events".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("options") {
        body.insert("options".to_string(), val.clone().into());
    }
    if !body.is_empty() {
        builder = builder.json(&serde_json::Value::Object(body));
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("POST /batch failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Set or update user properties without sending an event
///
/// Method: POST /identify
#[actor(
    AmplitudeUpdateUserPropertyActor,
    inports::<100>(api_key, identification, user_id, device_id, user_properties),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_update_user_property(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/identify".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.post(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut body = serde_json::Map::new();
    if let Some(val) = inputs.get("api_key") {
        body.insert("api_key".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("identification") {
        body.insert("identification".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("user_id") {
        body.insert("user_id".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("device_id") {
        body.insert("device_id".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("user_properties") {
        body.insert("user_properties".to_string(), val.clone().into());
    }
    if !body.is_empty() {
        builder = builder.json(&serde_json::Value::Object(body));
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("POST /identify failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Get metrics for an event with segmentation
///
/// Method: GET /events/segmentation
#[actor(
    AmplitudeReadEventSegmentationActor,
    inports::<100>(e, start, end, m, i, s, g, limit, p),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_event_segmentation(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/events/segmentation".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("e") {
        query_pairs.push(("e", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("start") {
        query_pairs.push(("start", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("end") {
        query_pairs.push(("end", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("m") {
        query_pairs.push(("m", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("i") {
        query_pairs.push(("i", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("s") {
        query_pairs.push(("s", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("g") {
        query_pairs.push(("g", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("limit") {
        query_pairs.push(("limit", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("p") {
        query_pairs.push(("p", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /events/segmentation failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Get funnel analysis data
///
/// Method: GET /funnels
#[actor(
    AmplitudeReadFunnelActor,
    inports::<100>(e, start, end, s, g, cvt, n),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_funnel(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/funnels".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("e") {
        query_pairs.push(("e", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("start") {
        query_pairs.push(("start", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("end") {
        query_pairs.push(("end", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("s") {
        query_pairs.push(("s", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("g") {
        query_pairs.push(("g", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("cvt") {
        query_pairs.push(("cvt", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("n") {
        query_pairs.push(("n", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /funnels failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Get user retention data
///
/// Method: GET /retention
#[actor(
    AmplitudeReadRetentionActor,
    inports::<100>(se, re, start, end, retention_type, s, n),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_retention(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/retention".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("se") {
        query_pairs.push(("se", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("re") {
        query_pairs.push(("re", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("start") {
        query_pairs.push(("start", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("end") {
        query_pairs.push(("end", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("retention_type") {
        query_pairs.push(("retention_type", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("s") {
        query_pairs.push(("s", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("n") {
        query_pairs.push(("n", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /retention failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Get a single user's event stream
///
/// Method: GET /useractivity
#[actor(
    AmplitudeReadUserActivityActor,
    inports::<100>(user, start, end, limit, offset),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_user_activity(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/useractivity".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("user") {
        query_pairs.push(("user", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("start") {
        query_pairs.push(("start", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("end") {
        query_pairs.push(("end", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("limit") {
        query_pairs.push(("limit", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("offset") {
        query_pairs.push(("offset", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /useractivity failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Find a user by user ID or device ID
///
/// Method: GET /usersearch
#[actor(
    AmplitudeReadUserSearchActor,
    inports::<100>(user),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_user_search(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/usersearch".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("user") {
        query_pairs.push(("user", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /usersearch failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// List all event types in the project
///
/// Method: GET /taxonomy/event
#[actor(
    AmplitudeReadEventsListActor,
    inports::<100>(limit, offset),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_events_list(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/taxonomy/event".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("limit") {
        query_pairs.push(("limit", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("offset") {
        query_pairs.push(("offset", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /taxonomy/event failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Get all properties for a specific event type
///
/// Method: GET /taxonomy/event/{eventName}/properties
#[actor(
    AmplitudeReadEventPropertiesActor,
    inports::<100>(eventName, limit, offset),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_event_properties(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let mut endpoint = "/taxonomy/event/{eventName}/properties".to_string();
    if let Some(val) = inputs.get("eventName") {
        endpoint = endpoint.replace("{{eventName}}", &super::message_to_str(val));
    }

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("limit") {
        query_pairs.push(("limit", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("offset") {
        query_pairs.push(("offset", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(
                    format!("GET /taxonomy/event/{{eventName}}/properties failed: {}", e).into(),
                ),
            );
        }
    }

    Ok(output)
}

/// List all user properties in the project
///
/// Method: GET /taxonomy/user-property
#[actor(
    AmplitudeReadUserPropertiesActor,
    inports::<100>(limit, offset),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_user_properties(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/taxonomy/user-property".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("limit") {
        query_pairs.push(("limit", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("offset") {
        query_pairs.push(("offset", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /taxonomy/user-property failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// List all cohorts in the project
///
/// Method: GET /cohorts
#[actor(
    AmplitudeReadCohortActor,
    inports::<100>(trigger),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_cohort(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/cohorts".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /cohorts failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Download a list of users in a cohort
///
/// Method: GET /cohorts/request/{cohortId}
#[actor(
    AmplitudeDownloadCohortMembersActor,
    inports::<100>(cohortId, props, propKeys),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_download_cohort_members(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let mut endpoint = "/cohorts/request/{cohortId}".to_string();
    if let Some(val) = inputs.get("cohortId") {
        endpoint = endpoint.replace("{{cohortId}}", &super::message_to_str(val));
    }

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("props") {
        query_pairs.push(("props", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("propKeys") {
        query_pairs.push(("propKeys", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /cohorts/request/{{cohortId}} failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// List all dashboards in the project
///
/// Method: GET /dashboards
#[actor(
    AmplitudeReadDashboardActor,
    inports::<100>(trigger),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_dashboard(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/dashboards".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /dashboards failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Delete all data for a user (GDPR/CCPA compliance)
///
/// Method: POST /deletions/users
#[actor(
    AmplitudeDeleteUserDataActor,
    inports::<100>(user_ids, amplitude_ids, requester, ignore_invalid_id, delete_from_org),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_delete_user_data(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/deletions/users".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.post(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut body = serde_json::Map::new();
    if let Some(val) = inputs.get("user_ids") {
        body.insert("user_ids".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("amplitude_ids") {
        body.insert("amplitude_ids".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("requester") {
        body.insert("requester".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("ignore_invalid_id") {
        body.insert("ignore_invalid_id".to_string(), val.clone().into());
    }
    if let Some(val) = inputs.get("delete_from_org") {
        body.insert("delete_from_org".to_string(), val.clone().into());
    }
    if !body.is_empty() {
        builder = builder.json(&serde_json::Value::Object(body));
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("POST /deletions/users failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Get status of user deletion jobs
///
/// Method: GET /deletions/users
#[actor(
    AmplitudeReadDeletionJobActor,
    inports::<100>(start_day, end_day),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_deletion_job(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/deletions/users".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("start_day") {
        query_pairs.push(("start_day", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("end_day") {
        query_pairs.push(("end_day", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /deletions/users failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}

/// Get lifetime value data per user cohort
///
/// Method: GET /revenue/ltv
#[actor(
    AmplitudeReadRevenueLtvActor,
    inports::<100>(start, end, m, g, s),
    outports::<50>(response, error),
    state(MemoryState)
)]
pub async fn amplitude_read_revenue_ltv(
    context: ActorContext,
) -> Result<HashMap<String, Message>, Error> {
    let inputs = context.get_payload();
    let actor_config = context.get_config();

    let endpoint = "/revenue/ltv".to_string();

    let url = format!("{}{}", BASE_URL.trim_end_matches('/'), endpoint);

    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;

    let mut builder = client.get(&url);
    builder = builder.header("Content-Type", "application/json");
    builder = apply_auth(actor_config, builder)?;

    let mut query_pairs: Vec<(&str, String)> = Vec::new();
    if let Some(val) = inputs.get("start") {
        query_pairs.push(("start", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("end") {
        query_pairs.push(("end", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("m") {
        query_pairs.push(("m", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("g") {
        query_pairs.push(("g", super::message_to_str(val)));
    }
    if let Some(val) = inputs.get("s") {
        query_pairs.push(("s", super::message_to_str(val)));
    }
    if !query_pairs.is_empty() {
        builder = builder.query(&query_pairs);
    }

    let mut output = HashMap::new();
    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let headers: HashMap<String, String> = resp
                .headers()
                .iter()
                .filter_map(|(k, v)| v.to_str().ok().map(|val| (k.to_string(), val.to_string())))
                .collect();
            let body_text = resp.text().await.unwrap_or_default();
            let body_value: Value =
                serde_json::from_str(&body_text).unwrap_or(Value::String(body_text));
            output.insert(
                "response".to_string(),
                Message::object(EncodableValue::from(json!({
                    "status": status,
                    "headers": headers,
                    "body": body_value,
                }))),
            );
        }
        Err(e) => {
            output.insert(
                "error".to_string(),
                Message::Error(format!("GET /revenue/ltv failed: {}", e).into()),
            );
        }
    }

    Ok(output)
}