// Copyright 2026 PingCAP, Inc.

use std::{collections::BTreeMap, fs, sync::Arc};

use async_trait::async_trait;
use aws::ActiveRefreshingProvider;
use chrono::{DateTime, Utc};
use hmac::{Hmac, Mac};
use hyper::{Body, Client, Request, client::HttpConnector, header::HeaderValue};
use hyper_tls::HttpsConnector;
use rusoto_credential::{AwsCredentials, CredentialsError, ProvideAwsCredentials};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use thiserror::Error;
use url::{Url, form_urlencoded::Serializer};

// `volces.com` matches both public `*.volces.com` and internal `*.ivolces.com`.
// Ref https://www.volcengine.com/docs/6349/147050
pub const DOMAIN_STRING: &str = "volces.com";

// Injected by VKE IRSA.
pub const ENV_VOLCENGINE_ROLE_TRN: &str = "VOLCENGINE_OIDC_ROLE_TRN";
pub const ENV_VOLCENGINE_OIDC_TOKEN_FILE: &str = "VOLCENGINE_OIDC_TOKEN_FILE";
pub const ENV_VOLCENGINE_REGION: &str = "REGION";

const STS_VERSION: &str = "2018-01-01";
const STS_ACTION_ASSUME_ROLE_WITH_OIDC: &str = "AssumeRoleWithOIDC";

const KMS_VERSION: &str = "2021-02-18";
const KMS_SERVICE: &str = "kms";
const KMS_ACTION_DECRYPT: &str = "Decrypt";

const DEFAULT_SESSION_NAME: &str = "TiKV";
const DEFAULT_DURATION_SECONDS: &str = "3600";

#[derive(Debug, Deserialize)]
struct VolcApiError {
    #[serde(rename = "Code")]
    code: Option<String>,
    #[serde(rename = "Message")]
    message: Option<String>,
}

#[derive(Debug, Deserialize)]
struct VolcResponseMetadata {
    #[serde(rename = "Error")]
    error: Option<VolcApiError>,
    #[serde(rename = "RequestId")]
    request_id: Option<String>,
}

#[derive(Debug, Deserialize)]
struct VolcResponse<T> {
    #[serde(rename = "ResponseMetadata")]
    response_metadata: VolcResponseMetadata,
    #[serde(rename = "Result")]
    result: Option<T>,
}

#[derive(Debug, Deserialize)]
struct AssumeRoleWithOidcResult {
    #[serde(rename = "Credentials")]
    credentials: Option<StsCredentials>,
}

#[derive(Debug, Deserialize)]
struct StsCredentials {
    #[serde(rename = "AccessKeyId")]
    access_key_id: String,
    #[serde(rename = "SecretAccessKey")]
    secret_access_key: String,
    #[serde(rename = "SessionToken")]
    session_token: String,
    #[serde(rename = "Expiration")]
    expiration: String,
}

#[derive(Debug, Serialize)]
struct DecryptRequest<'a> {
    #[serde(rename = "CiphertextBlob")]
    ciphertext_blob: &'a str,
    // Volcengine KMS wants EncryptionContext as a JSON-object-shaped string nested
    // inside the request body, not as a nested JSON object.
    #[serde(rename = "EncryptionContext", skip_serializing_if = "Option::is_none")]
    encryption_context: Option<&'a str>,
}

#[derive(Debug, Deserialize)]
struct DecryptResult {
    #[serde(rename = "Plaintext")]
    plaintext: Option<String>,
}

#[derive(Debug, Error)]
enum VolcengineError {
    #[error("http error: {0}")]
    Http(#[from] hyper::Error),
    #[error("invalid endpoint: {0}")]
    InvalidEndpoint(#[from] url::ParseError),
    #[error("invalid response: {0}")]
    InvalidResponse(String),
    #[error("serde error: {0}")]
    Serde(#[from] serde_json::Error),
    #[error("base64 error: {0}")]
    Base64(#[from] base64::DecodeError),
}

struct EndpointParts {
    request_url: String,
    // Used both as the literal `Host` header and as the canonical host in the
    // signature, so the two always agree.
    host_header: String,
    path: String,
}

pub struct AssumeRoleWithOidcProvider {
    region: String,
    role_trn: String,
    token_file: String,
    client: Client<HttpsConnector<HttpConnector>, Body>,
}

impl AssumeRoleWithOidcProvider {
    pub fn new() -> Result<Self, CredentialsError> {
        let region = read_region()?;
        let role_trn = read_env_required(ENV_VOLCENGINE_ROLE_TRN)?;
        let token_file = read_env_required(ENV_VOLCENGINE_OIDC_TOKEN_FILE)?;
        let client = Client::builder().build(HttpsConnector::new());
        Ok(Self {
            region,
            role_trn,
            token_file,
            client,
        })
    }
}

#[async_trait]
impl ProvideAwsCredentials for AssumeRoleWithOidcProvider {
    async fn credentials(&self) -> Result<AwsCredentials, CredentialsError> {
        let oidc_token = fs::read_to_string(&self.token_file)
            .map_err(|e| CredentialsError::new(format!("read oidc token file failed: {}", e)))?;
        let oidc_token = oidc_token.trim();
        if oidc_token.is_empty() {
            return Err(CredentialsError::new("oidc token is empty"));
        }
        let payload = build_form_encoded(&[
            ("DurationSeconds", DEFAULT_DURATION_SECONDS.to_owned()),
            ("OIDCToken", oidc_token.to_owned()),
            ("RoleSessionName", DEFAULT_SESSION_NAME.to_owned()),
            ("RoleTrn", self.role_trn.clone()),
        ]);
        let req = build_sts_request(&self.region, payload)
            .map_err(|e| CredentialsError::new(format!("build sts request failed: {}", e)))?;

        let resp = self
            .client
            .request(req)
            .await
            .map_err(|e| CredentialsError::new(format!("sts request failed: {}", e)))?;
        let response: VolcResponse<AssumeRoleWithOidcResult> = parse_response("sts", resp)
            .await
            .map_err(|e| CredentialsError::new(format!("parse sts response failed: {}", e)))?;
        let request_id_suffix = format_request_id(response.response_metadata.request_id.as_deref());
        if let Some(err) = response.response_metadata.error {
            return Err(CredentialsError::new(format!(
                "sts error {}: {}{}",
                err.code.unwrap_or_else(|| "Unknown".to_owned()),
                err.message.unwrap_or_else(|| "unknown error".to_owned()),
                request_id_suffix
            )));
        }
        let creds = response
            .result
            .and_then(|result| result.credentials)
            .ok_or_else(|| {
                CredentialsError::new(format!(
                    "sts response missing credentials field{}",
                    request_id_suffix
                ))
            })?;
        let expiration = DateTime::parse_from_rfc3339(creds.expiration.trim()).map_err(|e| {
            CredentialsError::new(format!("invalid Expiration: {}{}", e, request_id_suffix))
        })?;
        Ok(AwsCredentials::new(
            creds.access_key_id,
            creds.secret_access_key,
            Some(creds.session_token),
            Some(expiration.with_timezone(&Utc)),
        ))
    }
}

pub fn new_credential_provider() -> Result<ActiveRefreshingProvider, CredentialsError> {
    let provider = AssumeRoleWithOidcProvider::new()?;
    Ok(ActiveRefreshingProvider::new(Arc::new(provider)))
}

pub async fn decrypt_master_key(
    cipher_text: &str,
    region: &str,
    endpoint: &str,
    encryption_context: &BTreeMap<String, String>,
) -> Result<Vec<u8>, CredentialsError> {
    let provider = AssumeRoleWithOidcProvider::new()?;
    let region = if region.is_empty() {
        provider.region.clone()
    } else {
        region.to_owned()
    };
    let creds = provider.credentials().await?;
    let encryption_context_json = if encryption_context.is_empty() {
        None
    } else {
        Some(serde_json::to_string(encryption_context).map_err(|e| {
            CredentialsError::new(format!("serialize EncryptionContext failed: {}", e))
        })?)
    };
    let payload = DecryptRequest {
        ciphertext_blob: cipher_text,
        encryption_context: encryption_context_json.as_deref(),
    };
    let payload = serde_json::to_string(&payload)
        .map_err(|e| CredentialsError::new(format!("serialize kms request failed: {}", e)))?;
    let response = call_kms(&region, endpoint, &creds, payload)
        .await
        .map_err(|e| CredentialsError::new(format!("kms decrypt failed: {}", e)))?;
    let request_id_suffix = format_request_id(response.response_metadata.request_id.as_deref());
    if let Some(err) = response.response_metadata.error {
        return Err(CredentialsError::new(format!(
            "kms error {}: {}{}",
            err.code.unwrap_or_else(|| "Unknown".to_owned()),
            err.message.unwrap_or_else(|| "unknown error".to_owned()),
            request_id_suffix
        )));
    }
    let plaintext = response
        .result
        .and_then(|result| result.plaintext)
        .ok_or_else(|| {
            CredentialsError::new(format!(
                "kms response missing plaintext field{}",
                request_id_suffix
            ))
        })?;
    base64::decode(plaintext).map_err(|e| {
        CredentialsError::new(format!(
            "decode plaintext failed: {}{}",
            e, request_id_suffix
        ))
    })
}

async fn call_kms(
    region: &str,
    endpoint_override: &str,
    creds: &AwsCredentials,
    payload: String,
) -> Result<VolcResponse<DecryptResult>, VolcengineError> {
    let endpoint = if endpoint_override.is_empty() {
        build_kms_endpoint(region).0
    } else {
        endpoint_override.to_owned()
    };
    let req = build_signed_request(
        &endpoint,
        KMS_SERVICE,
        region,
        &build_query_string(&[
            ("Action", KMS_ACTION_DECRYPT.to_owned()),
            ("Version", KMS_VERSION.to_owned()),
        ]),
        "application/json",
        payload,
        creds.aws_access_key_id(),
        creds.aws_secret_access_key(),
        creds.token().as_deref(),
    )?;
    let client = Client::builder().build(HttpsConnector::new());
    let resp = client.request(req).await?;
    parse_response("kms", resp).await
}

async fn parse_response<T>(
    api: &str,
    resp: hyper::Response<Body>,
) -> Result<VolcResponse<T>, VolcengineError>
where
    T: for<'de> Deserialize<'de>,
{
    let status = resp.status();
    let body = hyper::body::to_bytes(resp.into_body()).await?;
    if !status.is_success() {
        return Err(VolcengineError::InvalidResponse(format!(
            "{} http error {}: {}{}",
            api,
            status,
            String::from_utf8_lossy(&body),
            format_request_id(None)
        )));
    }
    serde_json::from_slice(&body).map_err(Into::into)
}

fn build_sts_endpoint(region: &str) -> (String, String) {
    let host = format!("sts.{}.volcengineapi.com", region);
    let endpoint = format!("https://{}/", host);
    (endpoint, host)
}

fn build_kms_endpoint(region: &str) -> (String, String) {
    let host = format!("kms.{}.volcengineapi.com", region);
    let endpoint = format!("https://{}/", host);
    (endpoint, host)
}

fn build_sts_request(region: &str, payload: String) -> Result<Request<Body>, VolcengineError> {
    let (sts_endpoint, sts_host) = build_sts_endpoint(region);
    let query_string = build_query_string(&[
        ("Action", STS_ACTION_ASSUME_ROLE_WITH_OIDC.to_owned()),
        ("Version", STS_VERSION.to_owned()),
    ]);
    let request_url = format!("{}?{}", sts_endpoint, query_string);
    // AssumeRoleWithOIDC is unsigned: the OIDC token in the body is the credential.
    Request::post(request_url)
        .header("Accept", "application/json")
        .header(
            "Content-Type",
            "application/x-www-form-urlencoded; charset=utf-8",
        )
        .header("Host", sts_host.as_str())
        .body(Body::from(payload))
        .map_err(|e| VolcengineError::InvalidResponse(e.to_string()))
}

#[allow(clippy::too_many_arguments)]
fn build_signed_request(
    endpoint: &str,
    service: &str,
    region: &str,
    query: &str,
    content_type: &str,
    payload: String,
    access_key_id: &str,
    secret_access_key: &str,
    session_token: Option<&str>,
) -> Result<Request<Body>, VolcengineError> {
    let endpoint = parse_endpoint(endpoint, query)?;
    let x_date = Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
    let x_content_sha256 = sha256_hex(payload.as_bytes());
    let authorization = build_authorization(
        access_key_id,
        secret_access_key,
        service,
        region,
        &x_date,
        &endpoint.host_header,
        &endpoint.path,
        query,
        content_type,
        &x_content_sha256,
        session_token,
    );

    let mut req = Request::post(endpoint.request_url)
        .header("Accept", "application/json")
        .header("Content-Type", content_type)
        .header("Host", &endpoint.host_header)
        .header("X-Date", &x_date)
        .header("X-Content-Sha256", &x_content_sha256)
        .header("Authorization", authorization)
        .body(Body::from(payload))
        .map_err(|e| VolcengineError::InvalidResponse(e.to_string()))?;

    if let Some(token) = session_token.filter(|token| !token.is_empty()) {
        let header = HeaderValue::from_str(token)
            .map_err(|e| VolcengineError::InvalidResponse(e.to_string()))?;
        req.headers_mut().insert("X-Security-Token", header);
    }

    Ok(req)
}

fn parse_endpoint(endpoint: &str, query: &str) -> Result<EndpointParts, VolcengineError> {
    let mut url = Url::parse(endpoint)?;
    url.set_query(None);
    url.set_fragment(None);

    let host = url
        .host_str()
        .ok_or_else(|| VolcengineError::InvalidResponse("missing endpoint host".to_owned()))?;
    // `Url::port()` is `None` when the port equals the scheme default (the
    // parser normalizes it away), so any `Some` here is a non-default port
    // that must be kept in the `Host` header.
    let host_header = match url.port() {
        Some(port) => format!("{}:{}", host, port),
        None => host.to_owned(),
    };
    let path = if url.path().is_empty() {
        "/".to_owned()
    } else {
        url.path().to_owned()
    };
    let request_url = format!("{}?{}", url.as_str(), query);
    Ok(EndpointParts {
        request_url,
        host_header,
        path,
    })
}

#[allow(clippy::too_many_arguments)]
fn build_authorization(
    access_key_id: &str,
    secret_access_key: &str,
    service: &str,
    region: &str,
    x_date: &str,
    host: &str,
    path: &str,
    query: &str,
    content_type: &str,
    x_content_sha256: &str,
    session_token: Option<&str>,
) -> String {
    let mut headers = vec![
        ("content-type", content_type.to_owned()),
        ("host", host.to_owned()),
        ("x-content-sha256", x_content_sha256.to_owned()),
        ("x-date", x_date.to_owned()),
    ];
    if let Some(token) = session_token.filter(|token| !token.is_empty()) {
        headers.push(("x-security-token", token.to_owned()));
    }
    headers.sort_by(|lhs, rhs| lhs.0.cmp(rhs.0));

    let canonical_headers = headers
        .iter()
        .map(|(key, value)| format!("{}:{}\n", key, value.trim()))
        .collect::<String>();
    let signed_headers = headers
        .iter()
        .map(|(key, _)| *key)
        .collect::<Vec<_>>()
        .join(";");
    let canonical_request = [
        "POST".to_owned(),
        normalize_path(path),
        query.to_owned(),
        canonical_headers,
        signed_headers.clone(),
        x_content_sha256.to_owned(),
    ]
    .join("\n");
    let hashed_canonical_request = sha256_hex(canonical_request.as_bytes());
    let credential_scope = format!("{}/{}/{}/request", &x_date[..8], region, service);
    let string_to_sign = [
        "HMAC-SHA256".to_owned(),
        x_date.to_owned(),
        credential_scope.clone(),
        hashed_canonical_request,
    ]
    .join("\n");
    let signing_key = signing_key(secret_access_key, &x_date[..8], region, service);
    let signature = hex::encode(hmac_sha256(&signing_key, &string_to_sign));
    format!(
        "HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
        access_key_id, credential_scope, signed_headers, signature
    )
}

fn read_env_required(name: &str) -> Result<String, CredentialsError> {
    match std::env::var(name) {
        Ok(v) if !v.is_empty() => Ok(v),
        Ok(_) => Err(CredentialsError::new(format!("{} is empty", name))),
        Err(_) => Err(CredentialsError::new(format!("{} is missing", name))),
    }
}

fn read_region() -> Result<String, CredentialsError> {
    if let Ok(region) = std::env::var(ENV_VOLCENGINE_REGION) {
        if !region.is_empty() {
            return Ok(region);
        }
    }
    Err(CredentialsError::new(format!(
        "{} is missing",
        ENV_VOLCENGINE_REGION
    )))
}

fn build_query_string(params: &[(&str, String)]) -> String {
    build_form_encoded(params)
}

fn build_form_encoded(params: &[(&str, String)]) -> String {
    let mut params = params
        .iter()
        .map(|(key, value)| ((*key).to_owned(), value.to_owned()))
        .collect::<Vec<_>>();
    params.sort_by(|lhs, rhs| lhs.cmp(rhs));
    let mut serializer = Serializer::new(String::new());
    for (key, value) in params {
        serializer.append_pair(&key, &value);
    }
    serializer.finish().replace('+', "%20")
}

fn normalize_path(path: &str) -> String {
    if path.is_empty() {
        "/".to_owned()
    } else {
        path.split('/')
            .map(encode_path_fragment)
            .collect::<Vec<_>>()
            .join("/")
    }
}

fn encode_path_fragment(fragment: &str) -> String {
    let mut output = String::with_capacity(fragment.len());
    for byte in fragment.as_bytes() {
        if should_escape(*byte) {
            output.push('%');
            output.push(char::from(b"0123456789ABCDEF"[(byte >> 4) as usize]));
            output.push(char::from(b"0123456789ABCDEF"[(byte & 0x0F) as usize]));
        } else {
            output.push(char::from(*byte));
        }
    }
    output
}

fn should_escape(byte: u8) -> bool {
    !matches!(byte, b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~')
}

fn signing_key(secret_access_key: &str, date: &str, region: &str, service: &str) -> Vec<u8> {
    // Volcengine seeds the signing key directly from the secret access key,
    // unlike AWS SigV4 which prefixes it with "AWS4".
    let k_date = hmac_sha256(secret_access_key.as_bytes(), date);
    let k_region = hmac_sha256(&k_date, region);
    let k_service = hmac_sha256(&k_region, service);
    hmac_sha256(&k_service, "request")
}

fn hmac_sha256(key: &[u8], input: &str) -> Vec<u8> {
    let mut mac = Hmac::<Sha256>::new_from_slice(key).expect("hmac can take key of any size");
    mac.update(input.as_bytes());
    mac.finalize().into_bytes().to_vec()
}

fn sha256_hex(input: &[u8]) -> String {
    hex::encode(Sha256::digest(input))
}

fn format_request_id(request_id: Option<&str>) -> String {
    match request_id {
        Some(id) if !id.is_empty() => format!(", RequestId: {}", id),
        _ => ", RequestId: unknown".to_owned(),
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use super::{
        DecryptRequest, build_authorization, build_form_encoded, build_kms_endpoint,
        build_sts_endpoint, build_sts_request, parse_endpoint,
    };

    #[test]
    fn test_build_form_encoded() {
        let encoded = build_form_encoded(&[
            ("Policy", r#"{"Statement":"allow all"}"#.to_owned()),
            ("RoleTrn", "trn:iam::1234567890:role/test-role".to_owned()),
            ("OIDCToken", "token with spaces".to_owned()),
        ]);
        assert_eq!(
            encoded,
            "OIDCToken=token%20with%20spaces&Policy=%7B%22Statement%22%3A%22allow%20all%22%7D&RoleTrn=trn%3Aiam%3A%3A1234567890%3Arole%2Ftest-role"
        );
    }

    #[test]
    fn test_build_sts_endpoint() {
        let (endpoint, host) = build_sts_endpoint("cn-beijing");
        assert_eq!(endpoint, "https://sts.cn-beijing.volcengineapi.com/");
        assert_eq!(host, "sts.cn-beijing.volcengineapi.com");
    }

    #[test]
    fn test_build_kms_endpoint() {
        let (endpoint, host) = build_kms_endpoint("cn-beijing");
        assert_eq!(endpoint, "https://kms.cn-beijing.volcengineapi.com/");
        assert_eq!(host, "kms.cn-beijing.volcengineapi.com");
    }

    #[test]
    fn test_build_sts_request_is_unsigned() {
        let req = build_sts_request("cn-beijing", "OIDCToken=token".to_owned()).unwrap();
        assert_eq!(
            req.uri().to_string(),
            "https://sts.cn-beijing.volcengineapi.com/?Action=AssumeRoleWithOIDC&Version=2018-01-01"
        );
        assert_eq!(
            req.headers().get("host").unwrap().to_str().unwrap(),
            "sts.cn-beijing.volcengineapi.com"
        );
        assert!(req.headers().get("authorization").is_none());
        assert!(req.headers().get("x-date").is_none());
        assert!(req.headers().get("x-content-sha256").is_none());
    }

    #[test]
    fn test_build_authorization_signature() {
        // Expected signatures are computed by an independent implementation of
        // the Volcengine signing algorithm; they pin the exact algorithm, not
        // just the output shape.
        let auth = build_authorization(
            "AKLTtest",
            "c2VjcmV0",
            "kms",
            "cn-beijing",
            "20260323T140151Z",
            "kms.cn-beijing.volcengineapi.com",
            "/",
            "Action=Decrypt&Version=2021-02-18",
            "application/json",
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            None,
        );
        assert_eq!(
            auth,
            "HMAC-SHA256 Credential=AKLTtest/20260323/cn-beijing/kms/request, \
             SignedHeaders=content-type;host;x-content-sha256;x-date, \
             Signature=1fab8e1fcb692fe214645e8c8bc13df56680502f1c8926acf271fd4a74445a48"
        );
    }

    #[test]
    fn test_build_authorization_signature_with_session_token() {
        let auth = build_authorization(
            "AKLTtest",
            "c2VjcmV0",
            "kms",
            "cn-beijing",
            "20260323T140151Z",
            "kms.cn-beijing.volcengineapi.com",
            "/",
            "Action=Decrypt&Version=2021-02-18",
            "application/json",
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            Some("SESSIONTOKEN123"),
        );
        assert_eq!(
            auth,
            "HMAC-SHA256 Credential=AKLTtest/20260323/cn-beijing/kms/request, \
             SignedHeaders=content-type;host;x-content-sha256;x-date;x-security-token, \
             Signature=517db06ca764b0763691c6a1f9da61844c1bb58f2288de81355c564225a66687"
        );
    }

    #[test]
    fn test_parse_endpoint_default_port_is_omitted_from_host() {
        let parts = parse_endpoint(
            "https://kms.cn-beijing.volcengineapi.com/",
            "Action=Decrypt&Version=2021-02-18",
        )
        .unwrap();
        assert_eq!(parts.host_header, "kms.cn-beijing.volcengineapi.com");
        assert_eq!(parts.path, "/");
        assert_eq!(
            parts.request_url,
            "https://kms.cn-beijing.volcengineapi.com/?Action=Decrypt&Version=2021-02-18"
        );

        // An explicit default port is normalized away by the URL parser.
        let parts = parse_endpoint("https://example.com:443/path", "q=1").unwrap();
        assert_eq!(parts.host_header, "example.com");
    }

    #[test]
    fn test_parse_endpoint_non_default_port_is_kept_in_host() {
        let parts = parse_endpoint("https://example.com:8443/path", "q=1").unwrap();
        assert_eq!(parts.host_header, "example.com:8443");
        assert_eq!(parts.path, "/path");
        assert_eq!(parts.request_url, "https://example.com:8443/path?q=1");
    }

    #[test]
    fn test_decrypt_request_omits_encryption_context_when_absent() {
        let req = DecryptRequest {
            ciphertext_blob: "cGxhaW50ZXh0",
            encryption_context: None,
        };
        assert_eq!(
            serde_json::to_string(&req).unwrap(),
            r#"{"CiphertextBlob":"cGxhaW50ZXh0"}"#
        );
    }

    #[test]
    fn test_decrypt_request_serializes_encryption_context_as_json_string() {
        let context = BTreeMap::from([
            ("tenant".to_owned(), "acme".to_owned()),
            ("purpose".to_owned(), "cmek".to_owned()),
        ]);
        let context_json = serde_json::to_string(&context).unwrap();
        let req = DecryptRequest {
            ciphertext_blob: "cGxhaW50ZXh0",
            encryption_context: Some(&context_json),
        };
        // EncryptionContext must be a JSON string nested in the request body (a
        // string whose *value* happens to be JSON), not a nested JSON object,
        // per Volcengine's KMS Decrypt API contract.
        assert_eq!(
            serde_json::to_string(&req).unwrap(),
            r#"{"CiphertextBlob":"cGxhaW50ZXh0","EncryptionContext":"{\"purpose\":\"cmek\",\"tenant\":\"acme\"}"}"#
        );
    }
}