From 64eb5130713854ae64635c73a4ee017a26f343e9 Mon Sep 17 00:00:00 2001 From: YellowSnnowmann <167776381+YellowSnnowmann@users.noreply.github.com> Date: Wed, 1 Apr 2026 19:29:05 +0530 Subject: [PATCH] feat(billing, team): add billing and team management RPC functionality (#159) * feat(billing, team): add billing and team management RPC functionality - Introduced billing module with methods for fetching current plans, purchasing plans, creating portal sessions, and topping up credits. - Added team management module with methods for listing team members, creating invites, listing invites, removing members, and changing member roles. - Updated core registry to include new billing and team controllers and schemas, enhancing the overall functionality of the application. - Implemented comprehensive tests for billing and team RPC methods to ensure reliability and correctness. These additions improve the application's capabilities in managing billing and team functionalities, providing a more robust user experience. * refactor(billing, team): improve code readability and structure * fix(billing, team): enhance error handling and response structure - Improved error handling in to provide clearer error messages when reading response bodies. - Updated validation in to ensure is a finite number greater than zero. - Refined output schemas for billing and team functions to include more descriptive fields, enhancing API clarity. - Introduced a new function to standardize URL path construction, improving code maintainability. - Added tests to verify the correctness of new output structures and API path building. These changes enhance the robustness and usability of the billing and team management functionalities. * feat(billing, team): add gateway normalization and route redaction functionality - Introduced function to standardize payment gateway inputs, ensuring only valid options (stripe, coinbase) are accepted, with defaults and error handling. - Enhanced function to utilize the new gateway normalization logic, improving input validation. - Added function to obscure sensitive identifiers in API route paths, enhancing security in logging. - Implemented unit tests for both and to ensure correctness and reliability of the new features. These changes improve the robustness of billing operations and enhance security in route handling. --- src/core/all.rs | 6 + src/core/jsonrpc.rs | 210 +++++++++++ src/openhuman/billing/mod.rs | 12 + src/openhuman/billing/ops.rs | 311 ++++++++++++++++ src/openhuman/billing/schemas.rs | 346 ++++++++++++++++++ src/openhuman/mod.rs | 2 + src/openhuman/team/mod.rs | 7 + src/openhuman/team/ops.rs | 352 ++++++++++++++++++ src/openhuman/team/schemas.rs | 330 +++++++++++++++++ tests/json_rpc_e2e.rs | 598 ++++++++++++++++++++++++++++++- 10 files changed, 2170 insertions(+), 4 deletions(-) create mode 100644 src/openhuman/billing/mod.rs create mode 100644 src/openhuman/billing/ops.rs create mode 100644 src/openhuman/billing/schemas.rs create mode 100644 src/openhuman/team/mod.rs create mode 100644 src/openhuman/team/ops.rs create mode 100644 src/openhuman/team/schemas.rs diff --git a/src/core/all.rs b/src/core/all.rs index 731e1aa20..c92dd0e12 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -63,6 +63,8 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::workspace::all_workspace_registered_controllers()); controllers.extend(crate::openhuman::tools::all_tools_registered_controllers()); controllers.extend(crate::openhuman::memory::all_memory_registered_controllers()); + controllers.extend(crate::openhuman::billing::all_billing_registered_controllers()); + controllers.extend(crate::openhuman::team::all_team_registered_controllers()); controllers } @@ -91,6 +93,8 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::workspace::all_workspace_controller_schemas()); schemas.extend(crate::openhuman::tools::all_tools_controller_schemas()); schemas.extend(crate::openhuman::memory::all_memory_controller_schemas()); + schemas.extend(crate::openhuman::billing::all_billing_controller_schemas()); + schemas.extend(crate::openhuman::team::all_team_controller_schemas()); schemas } @@ -125,6 +129,8 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "skills" => Some("Skill registry, runtime lifecycle, setup, tools, and sync."), "socket" => Some("Skills runtime socket bridge controls."), "memory" => Some("Document storage, vector search, key-value store, and knowledge graph."), + "billing" => Some("Subscription plan, payment links, and credit top-up via the backend."), + "team" => Some("Team member management, invites, and role changes via the backend."), _ => None, } } diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 70f3c918c..f062f7814 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -732,5 +732,215 @@ mod tests { .any(|m| m.method == "openhuman.health_snapshot"), "schema dump should include migrated openhuman methods" ); + + assert!( + methods + .iter() + .any(|m| m.method == "openhuman.billing_get_current_plan"), + "schema dump should include billing methods" + ); + + assert!( + methods + .iter() + .any(|m| m.method == "openhuman.team_list_members"), + "schema dump should include team methods" + ); + } + + #[tokio::test] + async fn billing_get_current_plan_rejects_unknown_param() { + let err = invoke_method( + default_state(), + "openhuman.billing_get_current_plan", + json!({ "extra": true }), + ) + .await + .expect_err("unknown param should fail"); + assert!(err.contains("unknown param 'extra'")); + } + + #[tokio::test] + async fn billing_purchase_plan_missing_plan_fails_validation() { + let err = invoke_method( + default_state(), + "openhuman.billing_purchase_plan", + json!({}), + ) + .await + .expect_err("missing plan should fail"); + assert!(err.contains("missing required param 'plan'")); + } + + #[tokio::test] + async fn billing_top_up_missing_amount_fails_validation() { + let err = invoke_method(default_state(), "openhuman.billing_top_up", json!({})) + .await + .expect_err("missing amountUsd should fail"); + assert!(err.contains("missing required param 'amountUsd'")); + } + + #[tokio::test] + async fn billing_top_up_rejects_unknown_param() { + let err = invoke_method( + default_state(), + "openhuman.billing_top_up", + json!({ "amountUsd": 10.0, "unknownField": true }), + ) + .await + .expect_err("unknown param should fail"); + assert!(err.contains("unknown param 'unknownField'")); + } + + #[tokio::test] + async fn billing_create_portal_session_rejects_unknown_param() { + let err = invoke_method( + default_state(), + "openhuman.billing_create_portal_session", + json!({ "x": 1 }), + ) + .await + .expect_err("unknown param should fail"); + assert!(err.contains("unknown param 'x'")); + } + + #[tokio::test] + async fn team_list_members_missing_team_id_fails_validation() { + let err = invoke_method(default_state(), "openhuman.team_list_members", json!({})) + .await + .expect_err("missing teamId should fail"); + assert!(err.contains("missing required param 'teamId'")); + } + + #[tokio::test] + async fn team_list_members_rejects_unknown_param() { + let err = invoke_method( + default_state(), + "openhuman.team_list_members", + json!({ "teamId": "t1", "extra": true }), + ) + .await + .expect_err("unknown param should fail"); + assert!(err.contains("unknown param 'extra'")); + } + + #[tokio::test] + async fn team_create_invite_missing_team_id_fails_validation() { + let err = invoke_method(default_state(), "openhuman.team_create_invite", json!({})) + .await + .expect_err("missing teamId should fail"); + assert!(err.contains("missing required param 'teamId'")); + } + + #[tokio::test] + async fn team_remove_member_missing_required_params_fails_validation() { + let err = invoke_method( + default_state(), + "openhuman.team_remove_member", + json!({ "teamId": "t1" }), + ) + .await + .expect_err("missing userId should fail"); + assert!(err.contains("missing required param 'userId'")); + } + + #[tokio::test] + async fn team_change_member_role_missing_role_fails_validation() { + let err = invoke_method( + default_state(), + "openhuman.team_change_member_role", + json!({ "teamId": "t1", "userId": "u1" }), + ) + .await + .expect_err("missing role should fail"); + assert!(err.contains("missing required param 'role'")); + } + + #[tokio::test] + async fn billing_create_coinbase_charge_missing_plan_fails_validation() { + let err = invoke_method( + default_state(), + "openhuman.billing_create_coinbase_charge", + json!({}), + ) + .await + .expect_err("missing plan should fail"); + assert!(err.contains("missing required param 'plan'")); + } + + #[tokio::test] + async fn billing_create_coinbase_charge_rejects_unknown_param() { + let err = invoke_method( + default_state(), + "openhuman.billing_create_coinbase_charge", + json!({ "plan": "pro", "extra": true }), + ) + .await + .expect_err("unknown param should fail"); + assert!(err.contains("unknown param 'extra'")); + } + + #[tokio::test] + async fn team_list_invites_missing_team_id_fails_validation() { + let err = invoke_method(default_state(), "openhuman.team_list_invites", json!({})) + .await + .expect_err("missing teamId should fail"); + assert!(err.contains("missing required param 'teamId'")); + } + + #[tokio::test] + async fn team_list_invites_rejects_unknown_param() { + let err = invoke_method( + default_state(), + "openhuman.team_list_invites", + json!({ "teamId": "t1", "extra": true }), + ) + .await + .expect_err("unknown param should fail"); + assert!(err.contains("unknown param 'extra'")); + } + + #[tokio::test] + async fn team_revoke_invite_missing_team_id_fails_validation() { + let err = invoke_method(default_state(), "openhuman.team_revoke_invite", json!({})) + .await + .expect_err("missing teamId should fail"); + assert!(err.contains("missing required param 'teamId'")); + } + + #[tokio::test] + async fn team_revoke_invite_missing_invite_id_fails_validation() { + let err = invoke_method( + default_state(), + "openhuman.team_revoke_invite", + json!({ "teamId": "t1" }), + ) + .await + .expect_err("missing inviteId should fail"); + assert!(err.contains("missing required param 'inviteId'")); + } + + #[tokio::test] + async fn schema_dump_includes_new_billing_and_team_methods() { + let dump = build_http_schema_dump(); + let methods: Vec<&str> = dump.methods.iter().map(|m| m.method.as_str()).collect(); + for expected in &[ + "openhuman.billing_get_current_plan", + "openhuman.billing_purchase_plan", + "openhuman.billing_create_portal_session", + "openhuman.billing_top_up", + "openhuman.billing_create_coinbase_charge", + "openhuman.team_list_members", + "openhuman.team_create_invite", + "openhuman.team_list_invites", + "openhuman.team_revoke_invite", + "openhuman.team_remove_member", + "openhuman.team_change_member_role", + ] { + assert!( + methods.contains(expected), + "schema dump missing expected method: {expected}" + ); + } } } diff --git a/src/openhuman/billing/mod.rs b/src/openhuman/billing/mod.rs new file mode 100644 index 000000000..b058a8a6f --- /dev/null +++ b/src/openhuman/billing/mod.rs @@ -0,0 +1,12 @@ +//! Billing and payment RPC adapters that thin-wrap the hosted API. +//! +//! Exposes plan lookup, purchase flows, and credit top-ups through the +//! standard controller registry (`openhuman.billing_*`). + +mod ops; +mod schemas; + +pub use ops::*; +pub use schemas::{ + all_billing_controller_schemas, all_billing_registered_controllers, billing_schemas, +}; diff --git a/src/openhuman/billing/ops.rs b/src/openhuman/billing/ops.rs new file mode 100644 index 000000000..52e5979f9 --- /dev/null +++ b/src/openhuman/billing/ops.rs @@ -0,0 +1,311 @@ +//! Billing and payment RPC ops — thin adapters that call the hosted API. +//! +//! # Security +//! All methods require a valid app-session JWT stored via `auth_store_session`. +//! The JWT is sent as `Authorization: Bearer …` to the backend. +//! **No server-side authorization is replicated here**: the backend enforces plan +//! ownership, tenant isolation, and payment policy on every request. +//! Callers that lack a valid session or sufficient permissions receive a +//! backend 401/403 error surfaced verbatim as an RPC error string. +//! API keys / JWTs are never written to logs (only redacted status codes + paths). + +use log::debug; +use reqwest::{header::AUTHORIZATION, Client, Method, Url}; +use serde::Serialize; +use serde_json::{json, Value}; +use std::time::Duration; + +use crate::api::config::effective_api_url; +use crate::api::jwt::{bearer_authorization_value, get_session_token}; +use crate::openhuman::config::Config; +use crate::rpc::RpcOutcome; + +const LOG_PREFIX: &str = "[billing]"; + +fn build_client() -> Result { + Client::builder() + .use_rustls_tls() + .http1_only() + .timeout(Duration::from_secs(120)) + .connect_timeout(Duration::from_secs(15)) + .build() + .map_err(|e| format!("failed to build HTTP client: {e}")) +} + +fn resolve_base(config: &Config) -> Result { + let base = effective_api_url(&config.api_url); + Url::parse(base.trim()).map_err(|e| format!("invalid api_url '{}': {e}", base)) +} + +async fn authed_request( + client: &Client, + base: &Url, + token: &str, + method: Method, + path: &str, + body: Option, +) -> Result { + let url = base + .join(path.trim_start_matches('/')) + .map_err(|e| format!("build URL failed: {e}"))?; + + let mut req = client + .request(method.clone(), url.clone()) + .header(AUTHORIZATION, bearer_authorization_value(token)); + + if let Some(b) = body { + req = req.json(&b); + } + + let resp = req + .send() + .await + .map_err(|e| format!("request failed: {e}"))?; + let status = resp.status(); + + let text = resp + .text() + .await + .map_err(|e| format!("failed to read response body: {e}"))?; + + debug!("{LOG_PREFIX} {} {} -> {}", method, url, status); + + let raw: Value = serde_json::from_str(&text).unwrap_or_else(|_| Value::String(text.clone())); + if !status.is_success() { + let msg = raw + .as_object() + .and_then(|o| { + o.get("message") + .or_else(|| o.get("error")) + .or_else(|| o.get("detail")) + .and_then(|v| v.as_str()) + }) + .unwrap_or(&text); + return Err(format!( + "backend responded with {} for {}: {}", + status.as_u16(), + url.path(), + msg + )); + } + + unwrap_api_envelope(raw) +} + +fn unwrap_api_envelope(raw: Value) -> Result { + if let Some(obj) = raw.as_object() { + if let Some(success) = obj.get("success").and_then(|v| v.as_bool()) { + if !success { + let msg = obj + .get("message") + .or_else(|| obj.get("error")) + .and_then(|v| v.as_str()) + .unwrap_or("request unsuccessful"); + return Err(msg.to_string()); + } + } + if let Some(data) = obj.get("data") { + return Ok(data.clone()); + } + } + Ok(raw) +} + +fn require_token(config: &Config) -> Result { + get_session_token(config)? + .and_then(|v| { + let t = v.trim().to_string(); + if t.is_empty() { + None + } else { + Some(t) + } + }) + .ok_or_else(|| "no backend session token; run auth_store_session first".to_string()) +} + +async fn get_authed_value( + config: &Config, + method: Method, + path: &str, + body: Option, +) -> Result { + let token = require_token(config)?; + let client = build_client()?; + let base = resolve_base(config)?; + authed_request(&client, &base, &token, method, path, body).await +} + +pub async fn get_current_plan(config: &Config) -> Result, String> { + let data = get_authed_value(config, Method::GET, "/payments/stripe/currentPlan", None).await?; + Ok(RpcOutcome::single_log( + data, + "current plan fetched from backend", + )) +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct PurchasePlanBody<'a> { + plan: &'a str, +} + +pub async fn purchase_plan(config: &Config, plan: &str) -> Result, String> { + let plan = plan.trim(); + if plan.is_empty() { + return Err("plan is required".to_string()); + } + + let body = json!(PurchasePlanBody { plan }); + let data = get_authed_value( + config, + Method::POST, + "/payments/stripe/purchasePlan", + Some(body), + ) + .await?; + + Ok(RpcOutcome::single_log( + data, + "plan purchase session created", + )) +} + +pub async fn create_portal_session(config: &Config) -> Result, String> { + let data = get_authed_value(config, Method::POST, "/payments/stripe/portal", None).await?; + Ok(RpcOutcome::single_log( + data, + "customer portal session created", + )) +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct TopUpBody { + amount_usd: f64, + #[serde(default = "default_gateway")] + gateway: String, +} + +fn default_gateway() -> String { + "stripe".to_string() +} + +fn normalize_gateway(gateway: Option) -> Result { + let gateway = gateway + .as_deref() + .map(str::trim) + .filter(|g| !g.is_empty()) + .map(str::to_ascii_lowercase) + .unwrap_or_else(default_gateway); + + if !matches!(gateway.as_str(), "stripe" | "coinbase") { + return Err("gateway must be one of: stripe, coinbase".to_string()); + } + + Ok(gateway) +} + +pub async fn top_up_credits( + config: &Config, + amount_usd: f64, + gateway: Option, +) -> Result, String> { + if !amount_usd.is_finite() || amount_usd <= 0.0 { + return Err("amountUsd must be a finite number greater than 0".to_string()); + } + + let gateway = normalize_gateway(gateway)?; + let body = TopUpBody { + amount_usd, + gateway, + }; + + let data = get_authed_value( + config, + Method::POST, + "/payments/credits/top-up", + Some(json!(body)), + ) + .await?; + + Ok(RpcOutcome::single_log(data, "credit top-up initiated")) +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct CoinbaseChargeBody<'a> { + plan: &'a str, + interval: &'a str, +} + +/// Create a Coinbase Commerce charge (the "payment link" for crypto / annual billing). +/// Maps to `POST /payments/coinbase/charge` — matches `billingApi.createCoinbaseCharge`. +pub async fn create_coinbase_charge( + config: &Config, + plan: &str, + interval: Option, +) -> Result, String> { + let plan = plan.trim(); + if plan.is_empty() { + return Err("plan is required".to_string()); + } + + let interval_str = interval + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .unwrap_or("annual"); + + let body = json!(CoinbaseChargeBody { + plan, + interval: interval_str, + }); + + let data = get_authed_value( + config, + Method::POST, + "/payments/coinbase/charge", + Some(body), + ) + .await?; + + Ok(RpcOutcome::single_log( + data, + "Coinbase payment link created", + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn normalize_gateway_defaults_to_stripe() { + assert_eq!(normalize_gateway(None).unwrap(), "stripe"); + assert_eq!( + normalize_gateway(Some(" ".to_string())).unwrap(), + "stripe" + ); + } + + #[test] + fn normalize_gateway_accepts_supported_values_case_insensitively() { + assert_eq!( + normalize_gateway(Some(" Stripe ".to_string())).unwrap(), + "stripe" + ); + assert_eq!( + normalize_gateway(Some("COINBASE".to_string())).unwrap(), + "coinbase" + ); + } + + #[test] + fn normalize_gateway_rejects_unknown_values() { + assert_eq!( + normalize_gateway(Some("paypal".to_string())), + Err("gateway must be one of: stripe, coinbase".to_string()) + ); + } +} diff --git a/src/openhuman/billing/schemas.rs b/src/openhuman/billing/schemas.rs new file mode 100644 index 000000000..9c1f7ebeb --- /dev/null +++ b/src/openhuman/billing/schemas.rs @@ -0,0 +1,346 @@ +use serde::de::DeserializeOwned; +use serde::Deserialize; +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::config::rpc as config_rpc; +use crate::rpc::RpcOutcome; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct PurchasePlanParams { + plan: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct TopUpParams { + amount_usd: f64, + #[serde(default)] + gateway: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct CoinbaseChargeParams { + plan: String, + #[serde(default)] + interval: Option, +} + +pub fn all_billing_controller_schemas() -> Vec { + vec![ + billing_schemas("billing_get_current_plan"), + billing_schemas("billing_purchase_plan"), + billing_schemas("billing_create_portal_session"), + billing_schemas("billing_top_up"), + billing_schemas("billing_create_coinbase_charge"), + ] +} + +pub fn all_billing_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: billing_schemas("billing_get_current_plan"), + handler: handle_billing_get_current_plan, + }, + RegisteredController { + schema: billing_schemas("billing_purchase_plan"), + handler: handle_billing_purchase_plan, + }, + RegisteredController { + schema: billing_schemas("billing_create_portal_session"), + handler: handle_billing_create_portal_session, + }, + RegisteredController { + schema: billing_schemas("billing_top_up"), + handler: handle_billing_top_up, + }, + RegisteredController { + schema: billing_schemas("billing_create_coinbase_charge"), + handler: handle_billing_create_coinbase_charge, + }, + ] +} + +pub fn billing_schemas(function: &str) -> ControllerSchema { + match function { + "billing_get_current_plan" => ControllerSchema { + namespace: "billing", + function: "get_current_plan", + description: "Fetch current subscription plan and entitlements.", + inputs: vec![], + outputs: vec![json_output( + "plan", + "Current plan payload from backend /payments/stripe/currentPlan.", + )], + }, + "billing_purchase_plan" => ControllerSchema { + namespace: "billing", + function: "purchase_plan", + description: "Create Stripe Checkout session for a plan purchase.", + inputs: vec![required_string( + "plan", + "Plan identifier (backend contract).", + )], + outputs: vec![ + output_field( + "checkoutUrl", + TypeSchema::Option(Box::new(TypeSchema::String)), + "Stripe Checkout URL returned by /payments/stripe/purchasePlan.", + ), + output_field( + "sessionId", + TypeSchema::String, + "Stripe Checkout session ID returned by /payments/stripe/purchasePlan.", + ), + ], + }, + "billing_create_portal_session" => ControllerSchema { + namespace: "billing", + function: "create_portal_session", + description: "Create Stripe customer portal session.", + inputs: vec![], + outputs: vec![output_field( + "portalUrl", + TypeSchema::String, + "Stripe customer portal URL returned by /payments/stripe/portal.", + )], + }, + "billing_top_up" => ControllerSchema { + namespace: "billing", + function: "top_up", + description: "Initiate credit top-up via Stripe/Coinbase.", + inputs: vec![ + FieldSchema { + name: "amountUsd", + ty: TypeSchema::F64, + comment: "Top-up amount in USD.", + required: true, + }, + optional_string("gateway", "Payment gateway (stripe|coinbase)."), + ], + outputs: vec![ + output_field( + "url", + TypeSchema::String, + "Hosted payment URL returned by /payments/credits/top-up.", + ), + output_field( + "gatewayTransactionId", + TypeSchema::String, + "Gateway transaction identifier returned by /payments/credits/top-up.", + ), + output_field( + "amountUsd", + TypeSchema::F64, + "Top-up amount in USD returned by /payments/credits/top-up.", + ), + output_field( + "gateway", + TypeSchema::String, + "Payment gateway used for the top-up.", + ), + ], + }, + "billing_create_coinbase_charge" => ControllerSchema { + namespace: "billing", + function: "create_coinbase_charge", + description: "Create a Coinbase Commerce payment link for crypto / annual billing.", + inputs: vec![ + required_string("plan", "Plan tier (e.g. pro, enterprise)."), + optional_string("interval", "Billing interval; defaults to 'annual'."), + ], + outputs: vec![ + output_field( + "gatewayTransactionId", + TypeSchema::String, + "Coinbase Commerce charge identifier returned by /payments/coinbase/charge.", + ), + output_field( + "hostedUrl", + TypeSchema::String, + "Hosted Coinbase Commerce payment URL returned by /payments/coinbase/charge.", + ), + output_field( + "status", + TypeSchema::String, + "Coinbase charge status returned by /payments/coinbase/charge.", + ), + output_field( + "expiresAt", + TypeSchema::String, + "Charge expiration timestamp returned by /payments/coinbase/charge.", + ), + ], + }, + _ => ControllerSchema { + namespace: "billing", + function: "unknown", + description: "Unknown billing controller.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +fn handle_billing_get_current_plan(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + to_json(crate::openhuman::billing::get_current_plan(&config).await?) + }) +} + +fn handle_billing_purchase_plan(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json(crate::openhuman::billing::purchase_plan(&config, payload.plan.trim()).await?) + }) +} + +fn handle_billing_create_portal_session(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + to_json(crate::openhuman::billing::create_portal_session(&config).await?) + }) +} + +fn handle_billing_top_up(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json( + crate::openhuman::billing::top_up_credits(&config, payload.amount_usd, payload.gateway) + .await?, + ) + }) +} + +fn handle_billing_create_coinbase_charge(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json( + crate::openhuman::billing::create_coinbase_charge( + &config, + payload.plan.trim(), + payload.interval, + ) + .await?, + ) + }) +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} + +fn deserialize_params(params: Map) -> Result { + serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}")) +} + +fn required_string(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::String, + comment, + required: true, + } +} + +fn optional_string(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment, + required: false, + } +} + +fn json_output(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Json, + comment, + required: true, + } +} + +fn output_field(name: &'static str, ty: TypeSchema, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty, + comment, + required: true, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn schema_names_are_stable() { + let s = billing_schemas("billing_top_up"); + assert_eq!(s.namespace, "billing"); + assert_eq!(s.function, "top_up"); + } + + #[test] + fn controller_lists_match_lengths() { + assert_eq!( + all_billing_controller_schemas().len(), + all_billing_registered_controllers().len() + ); + } + + #[test] + fn schemas_match_unwrapped_backend_payload_keys() { + let purchase = billing_schemas("billing_purchase_plan"); + assert_eq!( + purchase + .outputs + .iter() + .map(|field| field.name) + .collect::>(), + vec!["checkoutUrl", "sessionId"] + ); + + let portal = billing_schemas("billing_create_portal_session"); + assert_eq!( + portal + .outputs + .iter() + .map(|field| field.name) + .collect::>(), + vec!["portalUrl"] + ); + + let top_up = billing_schemas("billing_top_up"); + assert_eq!( + top_up + .outputs + .iter() + .map(|field| field.name) + .collect::>(), + vec!["url", "gatewayTransactionId", "amountUsd", "gateway"] + ); + + let coinbase = billing_schemas("billing_create_coinbase_charge"); + assert_eq!( + coinbase + .outputs + .iter() + .map(|field| field.name) + .collect::>(), + vec!["gatewayTransactionId", "hostedUrl", "status", "expiresAt"] + ); + } +} diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index ad35a6091..184dd03bb 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -14,6 +14,7 @@ pub mod agent; pub mod approval; pub mod autocomplete; +pub mod billing; pub mod channels; pub mod config; pub mod cost; @@ -33,6 +34,7 @@ pub mod screen_intelligence; pub mod security; pub mod service; pub mod skills; +pub mod team; pub mod tools; pub mod util; pub mod webhooks; diff --git a/src/openhuman/team/mod.rs b/src/openhuman/team/mod.rs new file mode 100644 index 000000000..3da32b83c --- /dev/null +++ b/src/openhuman/team/mod.rs @@ -0,0 +1,7 @@ +//! Team management RPC adapters for member list and invites. + +mod ops; +mod schemas; + +pub use ops::*; +pub use schemas::{all_team_controller_schemas, all_team_registered_controllers, team_schemas}; diff --git a/src/openhuman/team/ops.rs b/src/openhuman/team/ops.rs new file mode 100644 index 000000000..b5286ded3 --- /dev/null +++ b/src/openhuman/team/ops.rs @@ -0,0 +1,352 @@ +//! Team management RPC ops — thin adapters that call the hosted API. +//! +//! # Security +//! All methods require a valid app-session JWT stored via `auth_store_session`. +//! The JWT is sent as `Authorization: Bearer …` to the backend. +//! **No server-side authorization is replicated here**: the backend enforces team +//! ownership, role permissions, and tenant isolation on every request. +//! Callers without the required role (e.g. non-owner trying to remove a member) +//! receive a backend 401/403 surfaced verbatim as an RPC error string. +//! API keys / JWTs are never written to logs. + +use log::debug; +use reqwest::{header::AUTHORIZATION, Client, Method, Url}; +use serde::Serialize; +use serde_json::{json, Value}; +use std::time::Duration; + +use crate::api::config::effective_api_url; +use crate::api::jwt::{bearer_authorization_value, get_session_token}; +use crate::openhuman::config::Config; +use crate::rpc::RpcOutcome; + +const LOG_PREFIX: &str = "[team]"; + +fn build_client() -> Result { + Client::builder() + .use_rustls_tls() + .http1_only() + .timeout(Duration::from_secs(120)) + .connect_timeout(Duration::from_secs(15)) + .build() + .map_err(|e| format!("failed to build HTTP client: {e}")) +} + +fn resolve_base(config: &Config) -> Result { + let base = effective_api_url(&config.api_url); + Url::parse(base.trim()).map_err(|e| format!("invalid api_url '{}': {e}", base)) +} + +fn require_token(config: &Config) -> Result { + get_session_token(config)? + .and_then(|v| { + let t = v.trim().to_string(); + if t.is_empty() { + None + } else { + Some(t) + } + }) + .ok_or_else(|| "no backend session token; run auth_store_session first".to_string()) +} + +fn normalize_id(input: &str, field: &str) -> Result { + let trimmed = input.trim(); + if trimmed.is_empty() { + return Err(format!("{field} is required")); + } + Ok(trimmed.to_string()) +} + +fn build_api_path(segments: &[&str]) -> Result { + let mut url = Url::parse("https://openhuman.invalid") + .map_err(|e| format!("failed to initialize URL path builder: {e}"))?; + { + let mut path_segments = url + .path_segments_mut() + .map_err(|_| "failed to initialize URL path builder".to_string())?; + path_segments.clear(); + for segment in segments { + path_segments.push(segment); + } + } + Ok(url.path().to_string()) +} + +fn is_identifier_segment(segment: &str) -> bool { + let trimmed = segment.trim(); + if trimmed.is_empty() { + return false; + } + + let allowed = |c: char| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '%' | '.'); + let has_digit = trimmed.chars().any(|c| c.is_ascii_digit()); + let is_uuid_like = trimmed.len() >= 8 + && trimmed.chars().all(allowed) + && trimmed.contains('-') + && trimmed.chars().any(|c| c.is_ascii_hexdigit()); + + (has_digit && trimmed.chars().all(allowed)) || is_uuid_like +} + +fn redact_route_template(url: &Url) -> String { + let Some(segments) = url.path_segments() else { + return url.path().to_string(); + }; + + let segments = segments.collect::>(); + let redacted = segments + .iter() + .enumerate() + .map(|(idx, segment)| { + match ( + idx, + segments.first().copied(), + segments.get(2).copied(), + *segment, + ) { + (1, Some("teams"), _, _) => "{team_id}".to_string(), + (3, Some("teams"), Some("members"), _) => "{user_id}".to_string(), + (3, Some("teams"), Some("invites"), _) => "{invite_id}".to_string(), + (_, _, _, value) + if is_identifier_segment(value) + && !matches!(value, "teams" | "members" | "invites" | "role") => + { + "{id}".to_string() + } + (_, _, _, value) => value.to_string(), + } + }) + .collect::>(); + + format!("/{}", redacted.join("/")) +} + +async fn authed_request( + client: &Client, + base: &Url, + token: &str, + method: Method, + path: &str, + body: Option, +) -> Result { + let url = base + .join(path.trim_start_matches('/')) + .map_err(|e| format!("build URL failed: {e}"))?; + + let mut req = client + .request(method.clone(), url.clone()) + .header(AUTHORIZATION, bearer_authorization_value(token)); + + if let Some(b) = body { + req = req.json(&b); + } + + let resp = req + .send() + .await + .map_err(|e| format!("request failed: {e}"))?; + let status = resp.status(); + let text = resp + .text() + .await + .map_err(|e| format!("failed to read backend response body: {e}"))?; + + debug!( + "{LOG_PREFIX} {} {} -> {}", + method, + redact_route_template(&url), + status + ); + + let raw: Value = serde_json::from_str(&text).unwrap_or_else(|_| Value::String(text.clone())); + if !status.is_success() { + let msg = raw + .as_object() + .and_then(|o| { + o.get("message") + .or_else(|| o.get("error")) + .or_else(|| o.get("detail")) + .and_then(|v| v.as_str()) + }) + .unwrap_or(&text); + return Err(format!( + "backend responded with {} for {}: {}", + status.as_u16(), + url.path(), + msg + )); + } + + unwrap_api_envelope(raw) +} + +fn unwrap_api_envelope(raw: Value) -> Result { + if let Some(obj) = raw.as_object() { + if let Some(success) = obj.get("success").and_then(|v| v.as_bool()) { + if !success { + let msg = obj + .get("message") + .or_else(|| obj.get("error")) + .and_then(|v| v.as_str()) + .unwrap_or("request unsuccessful"); + return Err(msg.to_string()); + } + } + if let Some(data) = obj.get("data") { + return Ok(data.clone()); + } + } + Ok(raw) +} + +async fn get_authed_value( + config: &Config, + method: Method, + path: &str, + body: Option, +) -> Result { + let token = require_token(config)?; + let client = build_client()?; + let base = resolve_base(config)?; + authed_request(&client, &base, &token, method, path, body).await +} + +pub async fn list_members(config: &Config, team_id: &str) -> Result, String> { + let team_id = normalize_id(team_id, "teamId")?; + let path = build_api_path(&["teams", &team_id, "members"])?; + let data = get_authed_value(config, Method::GET, &path, None).await?; + Ok(RpcOutcome::single_log( + data, + "team members fetched from backend", + )) +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct InviteBody { + #[serde(skip_serializing_if = "Option::is_none")] + max_uses: Option, + #[serde(skip_serializing_if = "Option::is_none")] + expires_in_days: Option, +} + +pub async fn create_invite( + config: &Config, + team_id: &str, + max_uses: Option, + expires_in_days: Option, +) -> Result, String> { + let team_id = normalize_id(team_id, "teamId")?; + let path = build_api_path(&["teams", &team_id, "invites"])?; + let body = json!(InviteBody { + max_uses, + expires_in_days, + }); + let data = get_authed_value(config, Method::POST, &path, Some(body)).await?; + Ok(RpcOutcome::single_log( + data, + "team invite created via backend", + )) +} + +pub async fn remove_member( + config: &Config, + team_id: &str, + user_id: &str, +) -> Result, String> { + let team_id = normalize_id(team_id, "teamId")?; + let user_id = normalize_id(user_id, "userId")?; + let path = build_api_path(&["teams", &team_id, "members", &user_id])?; + let data = get_authed_value(config, Method::DELETE, &path, None).await?; + Ok(RpcOutcome::single_log( + data, + "team member removed via backend", + )) +} + +pub async fn change_member_role( + config: &Config, + team_id: &str, + user_id: &str, + role: &str, +) -> Result, String> { + let team_id = normalize_id(team_id, "teamId")?; + let user_id = normalize_id(user_id, "userId")?; + let role = normalize_id(role, "role")?; + let path = build_api_path(&["teams", &team_id, "members", &user_id, "role"])?; + let body = json!({ "role": role }); + let data = get_authed_value(config, Method::PUT, &path, Some(body)).await?; + Ok(RpcOutcome::single_log( + data, + "team member role updated via backend", + )) +} + +/// List all active invites for a team. +/// Maps to `GET /teams/:teamId/invites` — matches `teamApi.getInvites`. +pub async fn list_invites(config: &Config, team_id: &str) -> Result, String> { + let team_id = normalize_id(team_id, "teamId")?; + let path = build_api_path(&["teams", &team_id, "invites"])?; + let data = get_authed_value(config, Method::GET, &path, None).await?; + Ok(RpcOutcome::single_log( + data, + "team invites listed from backend", + )) +} + +/// Revoke (delete) an existing invite by id. +/// Maps to `DELETE /teams/:teamId/invites/:inviteId` — matches `teamApi.revokeInvite`. +pub async fn revoke_invite( + config: &Config, + team_id: &str, + invite_id: &str, +) -> Result, String> { + let team_id = normalize_id(team_id, "teamId")?; + let invite_id = normalize_id(invite_id, "inviteId")?; + let path = build_api_path(&["teams", &team_id, "invites", &invite_id])?; + let data = get_authed_value(config, Method::DELETE, &path, None).await?; + Ok(RpcOutcome::single_log( + data, + "team invite revoked via backend", + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn build_api_path_encodes_reserved_characters_in_segments() { + let path = build_api_path(&["teams", "team/with?reserved", "members", "user#frag"]) + .expect("path should build"); + + assert_eq!(path, "/teams/team%2Fwith%3Freserved/members/user%23frag"); + } + + #[test] + fn redact_route_template_hides_team_member_and_invite_ids() { + let members_url = + Url::parse("https://api.example.test/teams/team-1/members").expect("members url"); + assert_eq!( + redact_route_template(&members_url), + "/teams/{team_id}/members" + ); + + let member_role_url = Url::parse( + "https://api.example.test/teams/69ca3f94bc6e00bbdc551900/members/user-2/role", + ) + .expect("member role url"); + assert_eq!( + redact_route_template(&member_role_url), + "/teams/{team_id}/members/{user_id}/role" + ); + + let invite_url = + Url::parse("https://api.example.test/teams/team-1/invites/inv-1").expect("invite url"); + assert_eq!( + redact_route_template(&invite_url), + "/teams/{team_id}/invites/{invite_id}" + ); + } +} diff --git a/src/openhuman/team/schemas.rs b/src/openhuman/team/schemas.rs new file mode 100644 index 000000000..7411c5401 --- /dev/null +++ b/src/openhuman/team/schemas.rs @@ -0,0 +1,330 @@ +use serde::de::DeserializeOwned; +use serde::Deserialize; +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::config::rpc as config_rpc; +use crate::rpc::RpcOutcome; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct TeamIdParams { + team_id: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct RemoveMemberParams { + team_id: String, + user_id: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct InviteParams { + team_id: String, + #[serde(default)] + max_uses: Option, + #[serde(default)] + expires_in_days: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ChangeRoleParams { + team_id: String, + user_id: String, + role: String, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct RevokeInviteParams { + team_id: String, + invite_id: String, +} + +pub fn all_team_controller_schemas() -> Vec { + vec![ + team_schemas("team_list_members"), + team_schemas("team_create_invite"), + team_schemas("team_list_invites"), + team_schemas("team_revoke_invite"), + team_schemas("team_remove_member"), + team_schemas("team_change_member_role"), + ] +} + +pub fn all_team_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: team_schemas("team_list_members"), + handler: handle_team_list_members, + }, + RegisteredController { + schema: team_schemas("team_create_invite"), + handler: handle_team_create_invite, + }, + RegisteredController { + schema: team_schemas("team_list_invites"), + handler: handle_team_list_invites, + }, + RegisteredController { + schema: team_schemas("team_revoke_invite"), + handler: handle_team_revoke_invite, + }, + RegisteredController { + schema: team_schemas("team_remove_member"), + handler: handle_team_remove_member, + }, + RegisteredController { + schema: team_schemas("team_change_member_role"), + handler: handle_team_change_member_role, + }, + ] +} + +pub fn team_schemas(function: &str) -> ControllerSchema { + match function { + "team_list_members" => ControllerSchema { + namespace: "team", + function: "list_members", + description: "List members for a team.", + inputs: vec![required_string("teamId", "Team id.")], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Raw member array returned by /teams/:teamId/members.", + required: true, + }], + }, + "team_create_invite" => ControllerSchema { + namespace: "team", + function: "create_invite", + description: "Create an invite for a team.", + inputs: vec![ + required_string("teamId", "Team id."), + optional_u64("maxUses", "Optional max uses."), + optional_u64("expiresInDays", "Optional expiry in days."), + ], + outputs: vec![json_output( + "result", + "Raw invite object returned by /teams/:teamId/invites.", + )], + }, + "team_remove_member" => ControllerSchema { + namespace: "team", + function: "remove_member", + description: "Remove a member from a team.", + inputs: vec![ + required_string("teamId", "Team id."), + required_string("userId", "User id to remove."), + ], + outputs: vec![json_output( + "result", + "Removal result payload from /teams/:teamId/members/:userId.", + )], + }, + "team_change_member_role" => ControllerSchema { + namespace: "team", + function: "change_member_role", + description: "Change a member's role in a team.", + inputs: vec![ + required_string("teamId", "Team id."), + required_string("userId", "User id."), + required_string("role", "Role identifier."), + ], + outputs: vec![json_output( + "result", + "Role update payload from /teams/:teamId/members/:userId/role.", + )], + }, + "team_list_invites" => ControllerSchema { + namespace: "team", + function: "list_invites", + description: "List active invites for a team.", + inputs: vec![required_string("teamId", "Team id.")], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Raw invite array returned by /teams/:teamId/invites.", + required: true, + }], + }, + "team_revoke_invite" => ControllerSchema { + namespace: "team", + function: "revoke_invite", + description: "Revoke (delete) an existing team invite.", + inputs: vec![ + required_string("teamId", "Team id."), + required_string("inviteId", "Invite id to revoke."), + ], + outputs: vec![json_output( + "result", + "Revoke result from /teams/:teamId/invites/:inviteId.", + )], + }, + _ => ControllerSchema { + namespace: "team", + function: "unknown", + description: "Unknown team controller.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +fn handle_team_list_members(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json(crate::openhuman::team::list_members(&config, &payload.team_id).await?) + }) +} + +fn handle_team_create_invite(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json( + crate::openhuman::team::create_invite( + &config, + &payload.team_id, + payload.max_uses, + payload.expires_in_days, + ) + .await?, + ) + }) +} + +fn handle_team_remove_member(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json( + crate::openhuman::team::remove_member(&config, &payload.team_id, &payload.user_id) + .await?, + ) + }) +} + +fn handle_team_change_member_role(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json( + crate::openhuman::team::change_member_role( + &config, + &payload.team_id, + &payload.user_id, + &payload.role, + ) + .await?, + ) + }) +} + +fn handle_team_list_invites(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json(crate::openhuman::team::list_invites(&config, &payload.team_id).await?) + }) +} + +fn handle_team_revoke_invite(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let payload = deserialize_params::(params)?; + to_json( + crate::openhuman::team::revoke_invite(&config, &payload.team_id, &payload.invite_id) + .await?, + ) + }) +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} + +fn deserialize_params(params: Map) -> Result { + serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}")) +} + +fn required_string(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::String, + comment, + required: true, + } +} + +fn optional_u64(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment, + required: false, + } +} + +fn json_output(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Json, + comment, + required: true, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn schema_names_are_stable() { + let s = team_schemas("team_list_members"); + assert_eq!(s.namespace, "team"); + assert_eq!(s.function, "list_members"); + } + + #[test] + fn controller_lists_match_lengths() { + assert_eq!( + all_team_controller_schemas().len(), + all_team_registered_controllers().len() + ); + } + + #[test] + fn schemas_match_unwrapped_team_payload_shapes() { + let members = team_schemas("team_list_members"); + assert_eq!(members.outputs.len(), 1); + assert_eq!(members.outputs[0].name, "result"); + assert_eq!( + members.outputs[0].ty, + TypeSchema::Array(Box::new(TypeSchema::Json)) + ); + + let create_invite = team_schemas("team_create_invite"); + assert_eq!(create_invite.outputs.len(), 1); + assert_eq!(create_invite.outputs[0].name, "result"); + assert_eq!(create_invite.outputs[0].ty, TypeSchema::Json); + + let invites = team_schemas("team_list_invites"); + assert_eq!(invites.outputs.len(), 1); + assert_eq!(invites.outputs[0].name, "result"); + assert_eq!( + invites.outputs[0].ty, + TypeSchema::Array(Box::new(TypeSchema::Json)) + ); + } +} diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index bec2e1d93..fd188563b 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -8,6 +8,7 @@ use std::path::Path; use std::sync::{Mutex, OnceLock}; use std::time::Duration; +use axum::http::{header::AUTHORIZATION, HeaderMap, StatusCode}; use axum::routing::{get, post}; use axum::{Json, Router}; use futures_util::StreamExt; @@ -60,15 +61,96 @@ fn json_rpc_e2e_env_lock() -> std::sync::MutexGuard<'static, ()> { } fn mock_upstream_router() -> Router { + const GENERAL_TOKEN: &str = "e2e-test-jwt"; + const BILLING_TOKEN: &str = "e2e-billing-jwt"; + const TEAM_TOKEN: &str = "e2e-team-jwt"; + + fn error_json(status: StatusCode, message: &str) -> (StatusCode, Json) { + ( + status, + Json(json!({ + "success": false, + "error": message, + "message": message, + })), + ) + } + + fn require_bearer( + headers: &HeaderMap, + expected_token: &str, + ) -> Result<(), (StatusCode, Json)> { + require_any_bearer(headers, &[expected_token]) + } + + fn require_any_bearer( + headers: &HeaderMap, + expected_tokens: &[&str], + ) -> Result<(), (StatusCode, Json)> { + let actual = headers + .get(AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .map(str::trim); + match actual { + Some(value) + if expected_tokens + .iter() + .any(|token| value == format!("Bearer {token}")) => + { + Ok(()) + } + Some(_) => Err(error_json( + StatusCode::UNAUTHORIZED, + "invalid Authorization bearer token", + )), + None => Err(error_json( + StatusCode::UNAUTHORIZED, + "missing Authorization bearer token", + )), + } + } + + fn require_string_field<'a>( + body: &'a Value, + field: &str, + ) -> Result<&'a str, (StatusCode, Json)> { + body.get(field) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .ok_or_else(|| { + error_json( + StatusCode::BAD_REQUEST, + &format!("missing or invalid '{field}'"), + ) + }) + } + + fn require_positive_f64_field( + body: &Value, + field: &str, + ) -> Result)> { + body.get(field) + .and_then(Value::as_f64) + .filter(|value| value.is_finite() && *value > 0.0) + .ok_or_else(|| { + error_json( + StatusCode::BAD_REQUEST, + &format!("missing or invalid '{field}'"), + ) + }) + } + // Matches `GET /settings` in `BackendOAuthClient::fetch_settings` (session store validation). - async fn settings() -> Json { - Json(json!({ + async fn settings(headers: HeaderMap) -> Result, (StatusCode, Json)> { + require_any_bearer(&headers, &[GENERAL_TOKEN, BILLING_TOKEN, TEAM_TOKEN])?; + Ok(Json(json!({ "success": true, "data": { "_id": "e2e-user-1", "username": "e2e" } - })) + }))) } async fn chat_completions(Json(_body): Json) -> Json { @@ -82,10 +164,238 @@ fn mock_upstream_router() -> Router { })) } + // ── Billing mock routes ────────────────────────────────────────────────── + + async fn stripe_current_plan( + headers: HeaderMap, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, BILLING_TOKEN)?; + Ok(Json(json!({ + "success": true, + "data": { + "plan": "PRO", + "hasActiveSubscription": true, + "planExpiry": "2030-01-01T00:00:00.000Z", + "subscription": { "id": "sub_mock_123", "status": "active" } + } + }))) + } + + async fn stripe_purchase_plan( + headers: HeaderMap, + Json(body): Json, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, BILLING_TOKEN)?; + let plan = require_string_field(&body, "plan")?; + if !matches!(plan, "basic" | "pro" | "BASIC" | "PRO") { + return Err(error_json( + StatusCode::BAD_REQUEST, + "missing or invalid 'plan'", + )); + } + + let checkout_url = "http://127.0.0.1/mock-checkout"; + let session_id = "cs_mock_abc"; + if checkout_url.is_empty() || session_id.is_empty() { + return Err(error_json( + StatusCode::BAD_REQUEST, + "missing checkoutUrl or sessionId", + )); + } + + Ok(Json(json!({ + "success": true, + "data": { "checkoutUrl": checkout_url, "sessionId": session_id } + }))) + } + + async fn stripe_portal(headers: HeaderMap) -> Result, (StatusCode, Json)> { + require_bearer(&headers, BILLING_TOKEN)?; + let portal_url = "http://127.0.0.1/mock-portal"; + if portal_url.is_empty() { + return Err(error_json(StatusCode::BAD_REQUEST, "missing portalUrl")); + } + + Ok(Json(json!({ + "success": true, + "data": { "portalUrl": portal_url } + }))) + } + + async fn credits_top_up( + headers: HeaderMap, + Json(body): Json, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, BILLING_TOKEN)?; + let amount_usd = require_positive_f64_field(&body, "amountUsd")?; + let gateway = require_string_field(&body, "gateway")?; + if !matches!(gateway, "stripe" | "coinbase") { + return Err(error_json( + StatusCode::BAD_REQUEST, + "missing or invalid 'gateway'", + )); + } + + Ok(Json(json!({ + "success": true, + "data": { + "url": "http://127.0.0.1/mock-topup", + "gatewayTransactionId": "txn_mock_1", + "amountUsd": amount_usd, + "gateway": gateway + } + }))) + } + + async fn coinbase_charge( + headers: HeaderMap, + Json(body): Json, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, BILLING_TOKEN)?; + let plan = require_string_field(&body, "plan")?; + let interval = body + .get("interval") + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or("annual"); + if !matches!(plan, "basic" | "pro" | "BASIC" | "PRO") { + return Err(error_json( + StatusCode::BAD_REQUEST, + "missing or invalid 'plan'", + )); + } + if interval != "annual" { + return Err(error_json( + StatusCode::BAD_REQUEST, + "missing or invalid 'interval'", + )); + } + + Ok(Json(json!({ + "success": true, + "data": { + "gatewayTransactionId": "coinbase_mock_1", + "hostedUrl": "http://127.0.0.1/mock-coinbase", + "status": "NEW", + "expiresAt": "2030-01-01T01:00:00.000Z" + } + }))) + } + + // ── Team mock routes ───────────────────────────────────────────────────── + + async fn team_members(headers: HeaderMap) -> Result, (StatusCode, Json)> { + require_bearer(&headers, TEAM_TOKEN)?; + Ok(Json(json!({ + "success": true, + "data": [ + { "id": "user-1", "username": "alice", "role": "ADMIN" }, + { "id": "user-2", "username": "bob", "role": "MEMBER" } + ] + }))) + } + + async fn team_invites_get( + headers: HeaderMap, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, TEAM_TOKEN)?; + Ok(Json(json!({ + "success": true, + "data": [ + { "id": "inv-1", "code": "ALPHA1", "maxUses": 5, "usedCount": 1, "expiresAt": null } + ] + }))) + } + + async fn team_invites_post( + headers: HeaderMap, + Json(body): Json, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, TEAM_TOKEN)?; + + let max_uses = body + .get("maxUses") + .and_then(Value::as_u64) + .ok_or_else(|| error_json(StatusCode::BAD_REQUEST, "missing or invalid 'maxUses'"))?; + let expires_in_days = body + .get("expiresInDays") + .and_then(Value::as_u64) + .ok_or_else(|| { + error_json( + StatusCode::BAD_REQUEST, + "missing or invalid 'expiresInDays'", + ) + })?; + if max_uses == 0 || expires_in_days == 0 { + return Err(error_json( + StatusCode::BAD_REQUEST, + "invite payload values must be greater than zero", + )); + } + + Ok(Json(json!({ + "success": true, + "data": { "id": "inv-new", "code": "NEWCODE", "maxUses": max_uses, "usedCount": 0, "expiresAt": null } + }))) + } + + async fn team_member_delete( + headers: HeaderMap, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, TEAM_TOKEN)?; + Ok(Json(json!({ "success": true, "data": {} }))) + } + + async fn team_member_role_put( + headers: HeaderMap, + Json(body): Json, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, TEAM_TOKEN)?; + let role = require_string_field(&body, "role")?; + if !matches!(role, "ADMIN" | "MEMBER" | "OWNER") { + return Err(error_json( + StatusCode::BAD_REQUEST, + "missing or invalid 'role'", + )); + } + Ok(Json(json!({ "success": true, "data": {} }))) + } + + async fn team_invite_delete( + headers: HeaderMap, + ) -> Result, (StatusCode, Json)> { + require_bearer(&headers, TEAM_TOKEN)?; + Ok(Json(json!({ "success": true, "data": {} }))) + } + Router::new() .route("/settings", get(settings)) - // `OpenHumanBackendProvider` uses `{api_url}/openai/v1` + `/chat/completions`. .route("/openai/v1/chat/completions", post(chat_completions)) + // billing + .route("/payments/stripe/currentPlan", get(stripe_current_plan)) + .route("/payments/stripe/purchasePlan", post(stripe_purchase_plan)) + .route("/payments/stripe/portal", post(stripe_portal)) + .route("/payments/credits/top-up", post(credits_top_up)) + .route("/payments/coinbase/charge", post(coinbase_charge)) + // team + .route("/teams/{team_id}/members", get(team_members)) + .route( + "/teams/{team_id}/members/{user_id}", + axum::routing::delete(team_member_delete), + ) + .route( + "/teams/{team_id}/members/{user_id}/role", + axum::routing::put(team_member_role_put), + ) + .route( + "/teams/{team_id}/invites", + get(team_invites_get).post(team_invites_post), + ) + .route( + "/teams/{team_id}/invites/{invite_id}", + axum::routing::delete(team_invite_delete), + ) } async fn serve_on_ephemeral( @@ -966,3 +1276,283 @@ async fn json_rpc_local_ai_device_profile_and_presets() { mock_join.abort(); rpc_join.abort(); } + +// ── Billing & Team E2E tests ────────────────────────────────────────────────── + +/// End-to-end test for billing RPC methods. +/// +/// Spins up an in-process Axum mock backend and a real JSON-RPC server, stores a +/// session JWT, then exercises every billing controller through the RPC surface +/// exactly as the desktop app or a CI script would. +#[tokio::test] +async fn billing_rpc_e2e() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; + let mock_origin = format!("http://{}", mock_addr); + write_min_config(&openhuman_home, &mock_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{}", rpc_addr); + + tokio::time::sleep(Duration::from_millis(100)).await; + + // Store a session first — all billing methods require it. + let store = post_json_rpc( + &rpc_base, + 1, + "openhuman.auth_store_session", + json!({ "token": "e2e-billing-jwt", "user_id": "e2e-user" }), + ) + .await; + assert_no_jsonrpc_error(&store, "store_session"); + + // Helper: the RPC outcome wraps backend data in {result: ..., logs: [...]}. + // We peel off the inner "result" field to get the actual backend payload. + fn inner(outer: &Value, _ctx: &str) -> Value { + outer + .get("result") + .cloned() + .unwrap_or_else(|| outer.clone()) + } + + // --- billing_get_current_plan --- + let plan = post_json_rpc( + &rpc_base, + 2, + "openhuman.billing_get_current_plan", + json!({}), + ) + .await; + let plan_outer = assert_no_jsonrpc_error(&plan, "billing_get_current_plan"); + let plan_result = inner(plan_outer, "billing_get_current_plan"); + assert_eq!( + plan_result.get("plan").and_then(Value::as_str), + Some("PRO"), + "expected PRO plan: {plan_result}" + ); + assert_eq!( + plan_result + .get("hasActiveSubscription") + .and_then(Value::as_bool), + Some(true), + "expected active subscription: {plan_result}" + ); + + // --- billing_purchase_plan --- + let purchase = post_json_rpc( + &rpc_base, + 3, + "openhuman.billing_purchase_plan", + json!({ "plan": "pro" }), + ) + .await; + let purchase_outer = assert_no_jsonrpc_error(&purchase, "billing_purchase_plan"); + let purchase_result = inner(purchase_outer, "billing_purchase_plan"); + assert!( + purchase_result + .get("checkoutUrl") + .and_then(Value::as_str) + .is_some(), + "expected checkoutUrl: {purchase_result}" + ); + + // --- billing_create_portal_session --- + let portal = post_json_rpc( + &rpc_base, + 4, + "openhuman.billing_create_portal_session", + json!({}), + ) + .await; + let portal_outer = assert_no_jsonrpc_error(&portal, "billing_create_portal_session"); + let portal_result = inner(portal_outer, "billing_create_portal_session"); + assert!( + portal_result + .get("portalUrl") + .and_then(Value::as_str) + .is_some(), + "expected portalUrl: {portal_result}" + ); + + // --- billing_top_up --- + let top_up = post_json_rpc( + &rpc_base, + 5, + "openhuman.billing_top_up", + json!({ "amountUsd": 10.0, "gateway": "stripe" }), + ) + .await; + let top_up_outer = assert_no_jsonrpc_error(&top_up, "billing_top_up"); + let top_up_result = inner(top_up_outer, "billing_top_up"); + assert_eq!( + top_up_result.get("amountUsd").and_then(Value::as_f64), + Some(10.0), + "expected amountUsd 10.0: {top_up_result}" + ); + + // --- billing_create_coinbase_charge --- + let charge = post_json_rpc( + &rpc_base, + 6, + "openhuman.billing_create_coinbase_charge", + json!({ "plan": "pro" }), + ) + .await; + let charge_outer = assert_no_jsonrpc_error(&charge, "billing_create_coinbase_charge"); + let charge_result = inner(charge_outer, "billing_create_coinbase_charge"); + assert!( + charge_result + .get("hostedUrl") + .and_then(Value::as_str) + .is_some(), + "expected hostedUrl: {charge_result}" + ); + assert_eq!( + charge_result.get("status").and_then(Value::as_str), + Some("NEW"), + "expected NEW status: {charge_result}" + ); + + mock_join.abort(); + rpc_join.abort(); +} + +/// End-to-end test for team RPC methods. +/// +/// Spins up an in-process Axum mock backend and a real JSON-RPC server, stores a +/// session JWT, then exercises every team controller through the RPC surface. +#[tokio::test] +async fn team_rpc_e2e() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; + let mock_origin = format!("http://{}", mock_addr); + write_min_config(&openhuman_home, &mock_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{}", rpc_addr); + + tokio::time::sleep(Duration::from_millis(100)).await; + + // Store a session first — all team methods require it. + let store = post_json_rpc( + &rpc_base, + 1, + "openhuman.auth_store_session", + json!({ "token": "e2e-team-jwt", "user_id": "e2e-user" }), + ) + .await; + assert_no_jsonrpc_error(&store, "store_session"); + + // Helper: peel off the inner "result" field from the RPC outcome envelope. + fn inner(outer: &Value, _ctx: &str) -> Value { + outer + .get("result") + .cloned() + .unwrap_or_else(|| outer.clone()) + } + + let team_id = "team-1"; + + // --- team_list_members --- + let members = post_json_rpc( + &rpc_base, + 2, + "openhuman.team_list_members", + json!({ "teamId": team_id }), + ) + .await; + let members_outer = assert_no_jsonrpc_error(&members, "team_list_members"); + let members_result = inner(members_outer, "team_list_members"); + let members_arr = members_result + .as_array() + .expect("expected array of members"); + assert_eq!(members_arr.len(), 2, "expected 2 members: {members_result}"); + assert_eq!( + members_arr[0].get("username").and_then(Value::as_str), + Some("alice") + ); + + // --- team_create_invite --- + let invite = post_json_rpc( + &rpc_base, + 3, + "openhuman.team_create_invite", + json!({ "teamId": team_id, "maxUses": 3, "expiresInDays": 7 }), + ) + .await; + let invite_outer = assert_no_jsonrpc_error(&invite, "team_create_invite"); + let invite_result = inner(invite_outer, "team_create_invite"); + assert!( + invite_result.get("code").and_then(Value::as_str).is_some(), + "expected invite code: {invite_result}" + ); + + // --- team_list_invites --- + let invites = post_json_rpc( + &rpc_base, + 4, + "openhuman.team_list_invites", + json!({ "teamId": team_id }), + ) + .await; + let invites_outer = assert_no_jsonrpc_error(&invites, "team_list_invites"); + let invites_result = inner(invites_outer, "team_list_invites"); + let invites_arr = invites_result + .as_array() + .expect("expected array of invites"); + assert!( + !invites_arr.is_empty(), + "expected at least one invite: {invites_result}" + ); + + // --- team_revoke_invite (no payload to check, just assert no error) --- + let revoke = post_json_rpc( + &rpc_base, + 5, + "openhuman.team_revoke_invite", + json!({ "teamId": team_id, "inviteId": "inv-1" }), + ) + .await; + assert_no_jsonrpc_error(&revoke, "team_revoke_invite"); + + // --- team_remove_member --- + let remove = post_json_rpc( + &rpc_base, + 6, + "openhuman.team_remove_member", + json!({ "teamId": team_id, "userId": "user-2" }), + ) + .await; + assert_no_jsonrpc_error(&remove, "team_remove_member"); + + // --- team_change_member_role --- + let role_change = post_json_rpc( + &rpc_base, + 7, + "openhuman.team_change_member_role", + json!({ "teamId": team_id, "userId": "user-1", "role": "MEMBER" }), + ) + .await; + assert_no_jsonrpc_error(&role_change, "team_change_member_role"); + + mock_join.abort(); + rpc_join.abort(); +}