openhuman/src/api/rest.rs

//! HTTP client for TinyHumans / AlphaHuman API routes (`/auth/...`, etc.).

use anyhow::{Context, Result};
use base64::Engine;
use reqwest::header::{HeaderMap, HeaderName, HeaderValue, AUTHORIZATION};
use reqwest::{Client, Method, Url};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::time::Duration;

use super::jwt::bearer_authorization_value;

/// Typed errors surfaced by `authed_json` for expected backend states that
/// callers should recover from in-flow rather than funnel into Sentry.
#[derive(Debug, thiserror::Error)]
pub enum BackendApiError {
    /// Edit / delete of a channel message returned 404. Happens when the
    /// user deletes the message on the provider side (Telegram, Discord,
    /// Slack, …) but our local `StreamingState` still has the id, or when
    /// the backend GC'd the relay row before we got around to editing it.
    /// Callers should clear stale state and skip the retry. Targets
    /// `OPENHUMAN-TAURI-2Y` (~454 events on `/channels/telegram/messages/<id>`).
    #[error("message not found on {provider}: {message_id}")]
    MessageNotFound {
        /// Channel provider segment (e.g. `"telegram"`, `"discord"`).
        provider: String,
        /// Provider-specific message id from the URL.
        message_id: String,
    },
    /// Backend rejected the bearer JWT with `401 Unauthorized`. This is an
    /// expected user-session state (token expired, revoked, rotated
    /// server-side) — not a code bug. Callers can route to a re-sign-in
    /// flow; the auth domain owns recovery. Targets `OPENHUMAN-TAURI-4K8`
    /// (12 events on `/openai/v1/audio/speech` mascot TTS, but the same
    /// shape fires on every authed endpoint once the session lapses).
    #[error("backend rejected session token on {method} {path}")]
    Unauthorized {
        /// HTTP method as a static string (`"GET"`, `"POST"`, …).
        method: String,
        /// Request path the 401 came back from (no query string).
        path: String,
    },
}

/// Extract `(provider, message_id)` from a backend channel path of the
/// shape `…/channels/<provider>/messages/<id>`. Returns `None` for paths
/// that do not contain this four-segment subsequence.
///
/// Handles both the canonical four-segment form and paths with an arbitrary
/// base-path prefix (e.g. `/api/v1/channels/telegram/messages/1103`) via a
/// sliding window so that `BACKEND_URL` variants with path prefixes do not
/// silently fall through to `report_error` (OPENHUMAN-TAURI-R7).
fn parse_message_path(path: &str) -> Option<(&str, &str)> {
    let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
    // Fast path: exact four-segment canonical form /channels/<p>/messages/<id>
    if segments.len() == 4 && segments[0] == "channels" && segments[2] == "messages" {
        return Some((segments[1], segments[3]));
    }
    // Sliding window: handles base-path prefixes like /api/v1/channels/<p>/messages/<id>
    for window in segments.windows(4) {
        if window[0] == "channels" && window[2] == "messages" {
            return Some((window[1], window[3]));
        }
    }
    None
}

const CLIENT_VERSION_HEADER_MAX_LEN: usize = 64;

fn sanitize_client_version(raw: &str) -> Option<String> {
    let sanitized: String = raw
        .trim()
        .chars()
        .filter(|c| matches!(c, '0'..='9' | 'A'..='Z' | 'a'..='z' | '.' | '_' | '+' | '-'))
        .take(CLIENT_VERSION_HEADER_MAX_LEN)
        .collect();

    if sanitized.is_empty() {
        None
    } else {
        Some(sanitized)
    }
}

fn build_backend_reqwest_client() -> Result<Client> {
    let mut default_headers = HeaderMap::new();
    if let Some(version) = sanitize_client_version(env!("CARGO_PKG_VERSION")) {
        default_headers.insert(
            HeaderName::from_static("x-core-version"),
            HeaderValue::from_str(&version).context("invalid x-core-version header value")?,
        );
    }
    // The Tauri shell sets `OPENHUMAN_TAURI_VERSION` to its own package version
    // before spawning the in-process core, so backend analytics can attribute
    // core-originated requests to the desktop shell build that hosts them.
    if let Ok(raw) = std::env::var("OPENHUMAN_TAURI_VERSION") {
        if let Some(version) = sanitize_client_version(&raw) {
            default_headers.insert(
                HeaderName::from_static("x-tauri-version"),
                HeaderValue::from_str(&version).context("invalid x-tauri-version header value")?,
            );
        }
    }

    // Platform-appropriate TLS backend: Windows → schannel (honors the OS
    // cert store, required for corporate TLS-inspection proxies); macOS /
    // Linux → rustls. See [`crate::openhuman::tls::tls_client_builder`].
    crate::openhuman::tls::tls_client_builder()
        .default_headers(default_headers)
        .http1_only()
        .timeout(Duration::from_secs(120))
        .connect_timeout(Duration::from_secs(15))
        .build()
        .map_err(|e| anyhow::anyhow!("failed to build HTTP client: {e}"))
}

fn parse_api_response_json(text: &str) -> Result<Value> {
    let v: Value = serde_json::from_str(text).with_context(|| format!("parse API JSON: {text}"))?;
    let Some(obj) = v.as_object() else {
        return Ok(v);
    };
    if let Some(success) = obj.get("success").and_then(|x| x.as_bool()) {
        if !success {
            let msg = obj
                .get("message")
                .or_else(|| obj.get("error"))
                .and_then(|x| x.as_str())
                .unwrap_or("request unsuccessful");
            anyhow::bail!("API request failed: {msg}");
        }
        if let Some(data) = obj.get("data") {
            if !data.is_null() {
                return Ok(data.clone());
            }
        }
        if let Some(user) = obj.get("user") {
            if !user.is_null() {
                return Ok(user.clone());
            }
        }
        let mut m = obj.clone();
        m.remove("success");
        return Ok(Value::Object(m));
    }
    Ok(v)
}

fn user_id_from_object(obj: &serde_json::Map<String, Value>) -> Option<String> {
    for key in ["id", "_id", "userId"] {
        if let Some(s) = obj.get(key).and_then(|x| x.as_str()) {
            let t = s.trim();
            if !t.is_empty() {
                return Some(t.to_string());
            }
        }
    }
    None
}

/// Best-effort extraction of a user ID from an authenticated profile payload.
///
/// This function handles various envelope formats, including raw user objects
/// or those nested under `data` or `user` keys.
pub fn user_id_from_profile_payload(payload: &Value) -> Option<String> {
    let obj = payload.as_object()?;
    if let Some(data) = obj.get("data").and_then(|v| v.as_object()) {
        return user_id_from_object(data).or_else(|| {
            data.get("user")
                .and_then(|u| u.as_object())
                .and_then(user_id_from_object)
        });
    }

    user_id_from_object(obj).or_else(|| {
        obj.get("user")
            .and_then(|u| u.as_object())
            .and_then(user_id_from_object)
    })
}

/// Alias for [`user_id_from_profile_payload`] for semantic clarity in auth flows.
pub fn user_id_from_auth_me_payload(payload: &Value) -> Option<String> {
    user_id_from_profile_payload(payload)
}

/// JSON body returned by the backend when an OAuth connection process is initiated.
#[derive(Debug, Clone, Deserialize)]
pub struct ConnectResponse {
    /// The URL to redirect the user to for OAuth authorization.
    pub oauth_url: String,
    /// The state parameter used to prevent CSRF and correlate the callback.
    pub state: String,
}

#[derive(Debug, Clone, Deserialize)]
struct ConnectEnvelope {
    success: bool,
    #[serde(default, alias = "oauthUrl")]
    oauth_url: Option<String>,
    #[serde(default)]
    state: Option<String>,
}

#[derive(Debug, Clone, Deserialize)]
struct IntegrationsEnvelope {
    success: bool,
    data: IntegrationsData,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct IntegrationsData {
    integrations: Vec<IntegrationSummary>,
}

/// A summary of an active integration, as returned by the backend.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct IntegrationSummary {
    /// Unique identifier for the integration.
    pub id: String,
    /// The name of the integration provider (e.g., "google", "slack").
    pub provider: String,
    /// RFC3339 timestamp of when the integration was created.
    pub created_at: String,
}

#[derive(Debug, Clone, Deserialize)]
struct TokensEnvelope {
    success: bool,
    data: TokensData,
}

#[derive(Debug, Clone, Deserialize)]
struct TokensData {
    encrypted: String,
}

#[derive(Debug, Clone, Deserialize)]
struct LoginTokenConsumeEnvelope {
    success: bool,
    data: LoginTokenConsumeData,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct LoginTokenConsumeData {
    jwt_token: String,
}

/// Decrypted OAuth token payload for handing off tokens to a local service or skill.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct IntegrationTokensHandoff {
    /// The OAuth access token.
    pub access_token: String,
    /// The optional OAuth refresh token.
    #[serde(default)]
    pub refresh_token: Option<String>,
    /// RFC3339 timestamp of when the access token expires.
    pub expires_at: String,
}

/// A client for interacting with the TinyHumans / AlphaHuman backend API.
#[derive(Clone)]
pub struct BackendOAuthClient {
    client: Client,
    base: Url,
}

impl BackendOAuthClient {
    /// Creates a new `BackendOAuthClient` with the given API base URL.
    ///
    /// Any path, query, or fragment in `api_base` is stripped so that
    /// `Url::join` always resolves root-relative REST paths correctly.
    /// This guards against callers who pass a full LLM completions URL
    /// (e.g. `https://host/v1/chat/completions`) instead of just the origin:
    /// without stripping, `join("teams/me/usage")` would produce the wrong
    /// path `/v1/chat/teams/me/usage` via RFC 3986 relative resolution.
    pub fn new(api_base: &str) -> Result<Self> {
        let mut base = Url::parse(api_base.trim()).context("Invalid API base URL")?;
        anyhow::ensure!(
            matches!(base.scheme(), "http" | "https") && base.host_str().is_some(),
            "API base URL must be an absolute http(s) URL with host"
        );
        base.set_path("");
        base.set_query(None);
        base.set_fragment(None);
        let client = build_backend_reqwest_client()?;
        Ok(Self { client, base })
    }

    /// Borrow the underlying `reqwest::Client` for callers that need to
    /// drive a non-JSON request shape (e.g. `multipart/form-data` uploads
    /// for cloud STT) without re-implementing TLS/proxy plumbing.
    pub fn raw_client(&self) -> &Client {
        &self.client
    }

    /// Resolve a backend-relative path against the configured base URL.
    /// Mirrors what `authed_json` does internally so callers using
    /// `raw_client()` don't have to assemble URLs by hand.
    pub fn url_for(&self, path: &str) -> Result<Url> {
        self.base
            .join(path.trim_start_matches('/'))
            .with_context(|| format!("build URL for {path}"))
    }

    /// Returns the URL for initiating a login flow for a specific provider.
    pub fn login_url(&self, provider: &str) -> Result<Url> {
        let p = provider.trim().trim_matches('/');
        anyhow::ensure!(!p.is_empty(), "provider is required");
        self.base
            .join(&format!("auth/{p}/login"))
            .context("build login URL")
    }

    /// Initiates an OAuth connection flow for the current user and a specific provider.
    pub async fn connect(
        &self,
        provider: &str,
        bearer_jwt: &str,
        skill_id: Option<&str>,
        response_type: Option<&str>,
        encryption_mode: Option<&str>,
    ) -> Result<ConnectResponse> {
        let p = provider.trim().trim_matches('/');
        anyhow::ensure!(!p.is_empty(), "provider is required");
        let mut url = self
            .base
            .join(&format!("auth/{p}/connect"))
            .context("build connect URL")?;
        if let Some(s) = skill_id.filter(|s| !s.is_empty()) {
            url.query_pairs_mut().append_pair("skillId", s);
        }
        if let Some(r) = response_type.filter(|r| !r.is_empty()) {
            url.query_pairs_mut().append_pair("responseType", r);
        }
        if let Some(e) = encryption_mode.filter(|e| !e.is_empty()) {
            url.query_pairs_mut().append_pair("encryptionMode", e);
        }

        let resp = self
            .client
            .get(url)
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt))
            .send()
            .await
            .context("auth connect request")?;

        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            anyhow::bail!("auth connect failed ({status}): {text}");
        }

        let env: ConnectEnvelope =
            serde_json::from_str(&text).with_context(|| format!("parse connect JSON: {text}"))?;
        if !env.success {
            anyhow::bail!("auth connect unsuccessful: {text}");
        }
        let oauth_url = env
            .oauth_url
            .filter(|u| !u.is_empty())
            .context("missing oauthUrl in response")?;
        let state = env
            .state
            .filter(|s| !s.is_empty())
            .context("missing state")?;
        Ok(ConnectResponse { oauth_url, state })
    }

    /// Fetches the current authenticated user profile using the provided JWT.
    pub async fn fetch_current_user(&self, bearer_jwt: &str) -> Result<Value> {
        let url = self.base.join("auth/me").context("build /auth/me URL")?;
        let resp = self
            .client
            .get(url)
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt))
            .send()
            .await
            .context("GET /auth/me")?;

        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            anyhow::bail!("GET /auth/me failed ({status}): {text}");
        }
        parse_api_response_json(&text)
    }

    /// Exchanges a one-time login token (e.g. from Telegram) for a long-lived JWT.
    pub async fn consume_login_token(&self, login_token: &str) -> Result<String> {
        let token = login_token.trim();
        anyhow::ensure!(!token.is_empty(), "login token is required");

        let url = self
            .base
            .join(&format!(
                "telegram/login-tokens/{}/consume",
                urlencoding::encode(token)
            ))
            .context("build login-token consume URL")?;

        let resp = self
            .client
            .post(url)
            .send()
            .await
            .context("consume login token")?;

        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            anyhow::bail!("consume login token failed ({status}): {text}");
        }

        let env: LoginTokenConsumeEnvelope = serde_json::from_str(&text)
            .with_context(|| format!("parse consume-login-token JSON: {text}"))?;
        if !env.success {
            anyhow::bail!("consume login token unsuccessful: {text}");
        }

        let jwt = env.data.jwt_token.trim().to_string();
        anyhow::ensure!(
            !jwt.is_empty(),
            "consume login token response missing jwtToken"
        );
        Ok(jwt)
    }

    /// Validates that the provided session token is still active and accepted.
    pub async fn validate_session_token(&self, bearer_jwt: &str) -> Result<()> {
        let _ = self.fetch_current_user(bearer_jwt).await?;
        Ok(())
    }

    /// Creates a short-lived link token for connecting a specific communication channel.
    pub async fn create_channel_link_token(
        &self,
        channel: &str,
        bearer_jwt: &str,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        let encoded_channel = urlencoding::encode(channel);

        let url = self
            .base
            .join(&format!("auth/channels/{encoded_channel}/link-token"))
            .context("build channel link-token URL")?;

        let resp = self
            .client
            .post(url)
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt))
            .send()
            .await
            .context("create channel link token")?;

        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            anyhow::bail!("create channel link token failed ({status}): {text}");
        }

        parse_api_response_json(&text)
    }

    /// Generic authenticated JSON request helper for backend API routes.
    pub async fn authed_json(
        &self,
        bearer_jwt: &str,
        method: Method,
        path: &str,
        body: Option<Value>,
    ) -> Result<Value> {
        let url = self
            .base
            .join(path.trim_start_matches('/'))
            .with_context(|| format!("build URL for {path}"))?;

        let mut request = self
            .client
            .request(method.clone(), url.clone())
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt));

        if let Some(body) = body {
            request = request.json(&body);
        }

        let response = request.send().await.map_err(|e| {
            // Walk the error source chain so transient markers hidden in nested
            // causes (reqwest -> hyper -> rustls TLS EOF, etc.) still classify
            // correctly. The top-level `e.to_string()` often only carries the
            // outermost wrapper, e.g. "error sending request for url (...)".
            let mut error_message = e.to_string();
            let mut src: Option<&(dyn std::error::Error + 'static)> = std::error::Error::source(&e);
            while let Some(s) = src {
                error_message.push_str(" → ");
                error_message.push_str(&s.to_string());
                src = s.source();
            }
            if crate::core::observability::contains_transient_transport_phrase(&error_message) {
                tracing::warn!(
                    domain = "backend_api",
                    operation = "authed_json",
                    method = method.as_str(),
                    path = url.path(),
                    failure = "transport",
                    error = %error_message,
                    "[backend_api] transient transport failure on {} {}: {}",
                    method.as_str(),
                    url.path(),
                    error_message,
                );
            } else {
                crate::core::observability::report_error(
                    error_message.as_str(),
                    "backend_api",
                    "authed_json",
                    &[
                        ("method", method.as_str()),
                        ("path", url.path()),
                        ("failure", "transport"),
                    ],
                );
            }
            anyhow::Error::new(e).context(format!(
                "backend request {} {}",
                method.as_str(),
                url.path()
            ))
        })?;

        let status = response.status();
        let text = response.text().await.unwrap_or_default();
        if !status.is_success() {
            let status_code = status.as_u16();
            let status_str = status_code.to_string();

            // 401 on any authed backend endpoint is an expected user-session
            // state (token expired / revoked / rotated server-side), not a
            // code bug — every authed endpoint will see this once the session
            // lapses. Surface a typed `BackendApiError::Unauthorized` so the
            // auth domain can drive recovery, and skip `report_error` to
            // avoid Sentry noise. Targets `OPENHUMAN-TAURI-4K8` (mascot TTS
            // surfaced it first on `/openai/v1/audio/speech`, but the same
            // shape applies to every `authed_json` path).
            if status_code == 401 {
                tracing::info!(
                    domain = "backend_api",
                    operation = "authed_json",
                    method = method.as_str(),
                    path = url.path(),
                    status = status_code,
                    failure = "non_2xx",
                    "[backend_api] 401 on {} {} — session token rejected, surfacing typed error",
                    method.as_str(),
                    url.path(),
                );
                return Err(anyhow::Error::new(BackendApiError::Unauthorized {
                    method: method.as_str().to_string(),
                    path: url.path().to_string(),
                }));
            }

            // 404 on `/channels/<provider>/messages/<id>` is an expected
            // state (user deleted the message provider-side, or backend
            // GC'd the relay row) — not a code bug. Surface a typed
            // `BackendApiError::MessageNotFound` so callers (`bus.rs`
            // streaming/thinking/delete/final paths) can clear stale
            // ids and skip retry, without funneling the 404 into
            // `report_error`. Targets `OPENHUMAN-TAURI-2Y` (~454 events).
            if status_code == 404 {
                if let Some((provider, message_id)) = parse_message_path(url.path()) {
                    tracing::info!(
                        domain = "backend_api",
                        operation = "authed_json",
                        provider = provider,
                        message_id = message_id,
                        "[backend_api] message-not-found 404 on {} {} — surfacing typed error",
                        method.as_str(),
                        url.path(),
                    );
                    return Err(anyhow::Error::new(BackendApiError::MessageNotFound {
                        provider: provider.to_string(),
                        message_id: message_id.to_string(),
                    }));
                }
                // Defense-in-depth: PATCH/DELETE 404s on any channel-message path that
                // parse_message_path could not parse (e.g. exotic URL variant with extra
                // segments). Still an expected backend state — suppress the Sentry event
                // without propagating a typed error. Targets OPENHUMAN-TAURI-R7.
                if (method == Method::PATCH || method == Method::DELETE)
                    && url.path().contains("/channels/")
                    && url.path().contains("/messages/")
                {
                    tracing::debug!(
                        domain = "backend_api",
                        operation = "authed_json",
                        "[backend_api] channel-message 404 on {} {} — path not matched by \
                         parse_message_path, suppressing Sentry (TAURI-R7 defense-in-depth)",
                        method.as_str(),
                        url.path(),
                    );
                    anyhow::bail!(
                        "channel message not found (404) on {} {}",
                        method.as_str(),
                        url.path(),
                    );
                }
            }

            // These are transient infrastructure errors (proxy/CDN/backend
            // temporarily unavailable). They are not code bugs and callers already
            // implement retry/disable logic, so skip Sentry to avoid noise.
            let is_transient_infra =
                crate::core::observability::is_transient_http_status_code(status_code);
            let is_budget_exhausted = status_code == 400
                && crate::openhuman::inference::provider::is_budget_exhausted_message(&text);
            if is_budget_exhausted {
                tracing::info!(
                    method = method.as_str(),
                    path = url.path(),
                    status = status_code,
                    failure = "non_2xx",
                    kind = "budget",
                    "[backend_api] budget-exhausted 400 on {} {} — not reporting to Sentry",
                    method.as_str(),
                    url.path(),
                );
            } else if is_transient_infra {
                tracing::warn!(
                    domain = "backend_api",
                    operation = "authed_json",
                    method = method.as_str(),
                    path = url.path(),
                    status = status_code,
                    failure = "non_2xx",
                    "[backend_api] transient {status} on {} {} — not reporting to Sentry",
                    method.as_str(),
                    url.path(),
                );
            } else {
                crate::core::observability::report_error(
                    format!(
                        "{} {} failed ({status}); response_body_len={}",
                        method.as_str(),
                        url.path(),
                        text.len()
                    )
                    .as_str(),
                    "backend_api",
                    "authed_json",
                    &[
                        ("method", method.as_str()),
                        ("path", url.path()),
                        ("status", status_str.as_str()),
                        ("failure", "non_2xx"),
                    ],
                );
            }
            anyhow::bail!(
                "{} {} failed ({status}): {text}",
                method.as_str(),
                url.path()
            );
        }

        parse_api_response_json(&text)
    }

    /// Lists all active integrations for the current user.
    pub async fn list_integrations(&self, bearer_jwt: &str) -> Result<Vec<IntegrationSummary>> {
        let url = self
            .base
            .join("auth/integrations")
            .context("build integrations URL")?;
        let resp = self
            .client
            .get(url)
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt))
            .send()
            .await
            .context("list integrations")?;

        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            anyhow::bail!("list integrations failed ({status}): {text}");
        }
        let env: IntegrationsEnvelope = serde_json::from_str(&text)
            .with_context(|| format!("parse integrations JSON: {text}"))?;
        if !env.success {
            anyhow::bail!("list integrations unsuccessful: {text}");
        }
        Ok(env.data.integrations)
    }

    /// Fetches the decrypted OAuth tokens for a specific integration.
    ///
    /// This is a one-time handoff process. The encryption key must match the
    /// one used by the backend to encrypt the tokens.
    pub async fn fetch_integration_tokens_handoff(
        &self,
        integration_id: &str,
        bearer_jwt: &str,
        encryption_key: &str,
    ) -> Result<IntegrationTokensHandoff> {
        let id = integration_id.trim();
        anyhow::ensure!(
            !id.is_empty() && id.len() == 24,
            "integrationId must be a 24-char hex id"
        );
        let url = self
            .base
            .join(&format!("auth/integrations/{id}/tokens"))
            .context("build tokens URL")?;
        let body = serde_json::json!({ "key": encryption_key.trim() });
        let resp = self
            .client
            .post(url)
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt))
            .json(&body)
            .send()
            .await
            .context("integration tokens handoff")?;

        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            anyhow::bail!("integration tokens failed ({status}): {text}");
        }
        let env: TokensEnvelope =
            serde_json::from_str(&text).with_context(|| format!("parse tokens JSON: {text}"))?;
        if !env.success {
            anyhow::bail!("integration tokens unsuccessful: {text}");
        }
        let plaintext = decrypt_handoff_blob(&env.data.encrypted, encryption_key.trim())?;
        serde_json::from_str(&plaintext).context("parse decrypted token JSON")
    }

    /// Fetches the client key share for a specific integration.
    ///
    /// This is a one-time handoff; the key is deleted from the backend's
    /// temporary storage (Redis) after retrieval.
    pub async fn fetch_client_key(&self, integration_id: &str, bearer_jwt: &str) -> Result<String> {
        let id = integration_id.trim();
        anyhow::ensure!(
            !id.is_empty() && id.len() == 24,
            "integrationId must be a 24-char hex id"
        );
        let url = self
            .base
            .join(&format!("auth/integrations/{id}/client-key"))
            .context("build client-key URL")?;
        let resp = self
            .client
            .post(url)
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt))
            .send()
            .await
            .context("fetch client key")?;

        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if !status.is_success() {
            anyhow::bail!("fetch client key failed ({status}): {text}");
        }
        let v: Value = serde_json::from_str(&text)
            .with_context(|| format!("parse client-key JSON: {text}"))?;
        let obj = v.as_object().context("expected JSON object")?;
        let success = obj
            .get("success")
            .and_then(|s| s.as_bool())
            .unwrap_or(false);
        if !success {
            let msg = obj
                .get("error")
                .and_then(|e| e.as_str())
                .unwrap_or("client key retrieval unsuccessful");
            anyhow::bail!("fetch client key failed: {msg}");
        }
        let client_key = obj
            .get("data")
            .and_then(|d| d.get("clientKey"))
            .and_then(|k| k.as_str())
            .context("missing data.clientKey in response")?;
        Ok(client_key.to_string())
    }

    /// Sends a message to a communication channel.
    pub async fn send_channel_message(
        &self,
        channel: &str,
        bearer_jwt: &str,
        message_body: Value,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        let encoded = urlencoding::encode(channel);
        self.authed_json(
            bearer_jwt,
            Method::POST,
            &format!("channels/{encoded}/messages"),
            Some(message_body),
        )
        .await
    }

    /// Signals "the agent is typing…" on a channel that supports it
    /// (Telegram's `sendChatAction`, Slack's typing event, …). The backend
    /// resolves the target chat from the channel integration metadata and
    /// is responsible for hitting the provider-native API.
    ///
    /// Telegram keeps the typing indicator alive for ~5 seconds per call,
    /// so callers should re-invoke every ~4 s for as long as the turn is
    /// in flight. Returns `Err` if the backend doesn't support typing for
    /// this channel — caller should swallow the error silently.
    pub async fn send_channel_typing(&self, channel: &str, bearer_jwt: &str) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        let encoded = urlencoding::encode(channel);
        self.authed_json(
            bearer_jwt,
            Method::POST,
            &format!("channels/{encoded}/typing"),
            Some(json!({})),
        )
        .await
    }

    /// Edits an existing channel message. Used by the progressive-edit
    /// streaming path (Telegram / Slack) to coalesce live deltas into a
    /// single evolving outbound message rather than spamming the chat
    /// with one bubble per token.
    ///
    /// `message_id` is the backend-returned id of the message that was
    /// first sent via [`Self::send_channel_message`]. Returns the
    /// updated message record, or an `Err` if the backend does not
    /// support editing for this channel (caller should fall back to
    /// atomic-final delivery).
    pub async fn send_channel_edit(
        &self,
        channel: &str,
        message_id: &str,
        bearer_jwt: &str,
        edit_body: Value,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        anyhow::ensure!(!message_id.is_empty(), "message_id is required");
        let encoded_channel = urlencoding::encode(channel);
        let encoded_id = urlencoding::encode(message_id);
        self.authed_json(
            bearer_jwt,
            Method::PATCH,
            &format!("channels/{encoded_channel}/messages/{encoded_id}"),
            Some(edit_body),
        )
        .await
    }

    /// Deletes a message from a communication channel. Used to clean up
    /// ephemeral messages (e.g. thinking indicators) after the final
    /// response has been delivered.
    pub async fn send_channel_delete(
        &self,
        channel: &str,
        message_id: &str,
        bearer_jwt: &str,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        anyhow::ensure!(!message_id.is_empty(), "message_id is required");
        let encoded_channel = urlencoding::encode(channel);
        let encoded_id = urlencoding::encode(message_id);
        self.authed_json(
            bearer_jwt,
            Method::DELETE,
            &format!("channels/{encoded_channel}/messages/{encoded_id}"),
            None,
        )
        .await
    }

    /// Sends a reaction (e.g. emoji) to a message in a channel.
    pub async fn send_channel_reaction(
        &self,
        channel: &str,
        bearer_jwt: &str,
        reaction_body: Value,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        let encoded = urlencoding::encode(channel);
        self.authed_json(
            bearer_jwt,
            Method::POST,
            &format!("channels/{encoded}/reactions"),
            Some(reaction_body),
        )
        .await
    }

    /// Creates a new thread in a communication channel.
    pub async fn create_channel_thread(
        &self,
        channel: &str,
        bearer_jwt: &str,
        title: &str,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        anyhow::ensure!(!title.trim().is_empty(), "title is required");
        let encoded = urlencoding::encode(channel);
        let body = serde_json::json!({ "title": title.trim() });
        self.authed_json(
            bearer_jwt,
            Method::POST,
            &format!("channels/{encoded}/threads"),
            Some(body),
        )
        .await
    }

    /// Updates an existing thread (e.g., closing or reopening it).
    pub async fn update_channel_thread(
        &self,
        channel: &str,
        bearer_jwt: &str,
        thread_id: &str,
        action: &str,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        anyhow::ensure!(!thread_id.trim().is_empty(), "threadId is required");
        anyhow::ensure!(
            action == "close" || action == "reopen",
            "action must be 'close' or 'reopen'"
        );
        let encoded_channel = urlencoding::encode(channel);
        let encoded_thread = urlencoding::encode(thread_id.trim());
        let body = serde_json::json!({ "action": action });
        self.authed_json(
            bearer_jwt,
            Method::PATCH,
            &format!("channels/{encoded_channel}/threads/{encoded_thread}"),
            Some(body),
        )
        .await
    }

    /// Lists threads in a communication channel, optionally filtering by status.
    pub async fn list_channel_threads(
        &self,
        channel: &str,
        bearer_jwt: &str,
        active_filter: Option<bool>,
    ) -> Result<Value> {
        let channel = channel.trim().trim_matches('/');
        anyhow::ensure!(!channel.is_empty(), "channel is required");
        let encoded = urlencoding::encode(channel);
        let mut path = format!("channels/{encoded}/threads");
        if let Some(active) = active_filter {
            path.push_str(if active {
                "?active=true"
            } else {
                "?active=false"
            });
        }
        self.authed_json(bearer_jwt, Method::GET, &path, None).await
    }

    /// Revokes (deletes) an active integration.
    pub async fn revoke_integration(&self, integration_id: &str, bearer_jwt: &str) -> Result<()> {
        let id = integration_id.trim();
        anyhow::ensure!(!id.is_empty(), "integration id is required");
        let url = self
            .base
            .join(&format!("auth/integrations/{id}"))
            .context("build revoke URL")?;
        let resp = self
            .client
            .delete(url)
            .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt))
            .send()
            .await
            .context("revoke integration")?;

        let status = resp.status();
        if !status.is_success() {
            let text = resp.text().await.unwrap_or_default();
            anyhow::bail!("revoke integration failed ({status}): {text}");
        }
        Ok(())
    }
}

/// AES-256-GCM decrypt compatible with backend `encryptMessageFromString` (IV 16 + tag 16 + ciphertext, base64).
pub fn decrypt_handoff_blob(b64_ciphertext: &str, key_str: &str) -> Result<String> {
    let key = key_bytes_from_string(key_str)?;
    let combined = base64::engine::general_purpose::STANDARD
        .decode(b64_ciphertext.trim())
        .context("base64-decode encrypted payload")?;
    if combined.len() < 32 {
        anyhow::bail!("encrypted payload too short");
    }
    let iv = &combined[0..16];
    let tag = &combined[16..32];
    let ciphertext = &combined[32..];

    // aes-gcm expects ciphertext || tag
    let mut ct_with_tag = Vec::with_capacity(ciphertext.len() + tag.len());
    ct_with_tag.extend_from_slice(ciphertext);
    ct_with_tag.extend_from_slice(tag);

    use aes_gcm::aead::generic_array::typenum::U16;
    use aes_gcm::aead::{Aead, KeyInit};
    use aes_gcm::aes::Aes256;
    use aes_gcm::AesGcm;
    type Aes256Gcm16 = AesGcm<Aes256, U16>;

    let cipher =
        Aes256Gcm16::new_from_slice(&key).map_err(|e| anyhow::anyhow!("invalid AES key: {e}"))?;
    let nonce = aes_gcm::aead::generic_array::GenericArray::from_slice(iv);
    let plain = cipher
        .decrypt(nonce, ct_with_tag.as_ref())
        .map_err(|e| anyhow::anyhow!("AES-GCM decrypt failed: {e}"))?;

    String::from_utf8(plain).context("handoff plaintext is not UTF-8")
}

/// Decode the shared encryption key into 32 raw AES bytes.
///
/// Accepts, in order of preference:
/// 1. base64url without padding — the current backend format (e.g.
///    a 43-char alphanumeric string using `-` / `_`). This must be tried
///    BEFORE standard base64 because `-`/`_` are invalid in the standard
///    alphabet and would fail cleanly, whereas a standard-base64 string
///    never contains `-`/`_` so base64url_no_pad will still decode it
///    correctly as long as there's no padding.
/// 2. base64url with padding.
/// 3. Standard base64 with padding (legacy backend format).
/// 4. Standard base64 without padding.
/// 5. A raw 32-byte ASCII key (len == 32, used as-is).
///
/// NOTE: the key is only decoded locally for AES-GCM key material in
/// `decrypt_handoff_blob`. The key sent back to the backend (in the
/// `{ key: ... }` POST body of `fetch_integration_tokens_handoff`) is the
/// original string — never re-encoded — so base64url keys stay base64url
/// on the wire.
fn key_bytes_from_string(key: &str) -> Result<Vec<u8>> {
    let trimmed = key.trim();

    // Raw 32-byte ASCII key
    if trimmed.len() == 32 && !trimmed.contains(['+', '/', '-', '_', '=']) {
        return Ok(trimmed.as_bytes().to_vec());
    }

    use base64::engine::general_purpose::{STANDARD, STANDARD_NO_PAD, URL_SAFE, URL_SAFE_NO_PAD};

    // `base64::Engine` has generic methods and therefore isn't
    // dyn-compatible, so we unroll the attempts instead of looping over
    // a slice of trait objects.
    macro_rules! try_decode {
        ($engine:expr) => {
            if let Ok(decoded) = $engine.decode(trimmed) {
                if decoded.len() == 32 {
                    return Ok(decoded);
                }
            }
        };
    }
    try_decode!(URL_SAFE_NO_PAD);
    try_decode!(URL_SAFE);
    try_decode!(STANDARD);
    try_decode!(STANDARD_NO_PAD);

    anyhow::bail!(
        "encryption key must decode to 32 raw bytes (raw, base64, or base64url accepted; got len={})",
        trimmed.len()
    );
}

#[cfg(test)]
#[path = "rest_tests.rs"]
mod key_bytes_from_string_tests;