Genarrative/server-rs/crates/platform-wechat/src/subscribe_message.rs

use std::{
    error::Error,
    fmt,
    sync::Arc,
    time::{Duration, Instant},
};

use reqwest::Client;
use ring::hmac;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tokio::sync::Mutex;
use tracing::warn;
use url::Url;

pub const DEFAULT_WECHAT_STABLE_ACCESS_TOKEN_ENDPOINT: &str =
    "https://api.weixin.qq.com/cgi-bin/stable_token";
pub const DEFAULT_WECHAT_VIRTUAL_PAYMENT_QUERY_ORDER_ENDPOINT: &str =
    "https://api.weixin.qq.com/xpay/query_order";
pub const DEFAULT_WECHAT_VIRTUAL_PAYMENT_NOTIFY_PROVIDE_GOODS_ENDPOINT: &str =
    "https://api.weixin.qq.com/xpay/notify_provide_goods";
const WECHAT_VIRTUAL_PAYMENT_QUERY_ORDER_URI: &str = "/xpay/query_order";
const WECHAT_ACCESS_TOKEN_REFRESH_SAFETY_MARGIN: Duration = Duration::from_secs(5 * 60);
const WECHAT_ACCESS_TOKEN_ERROR_CODES: [i64; 3] = [40001, 40014, 42001];

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WechatConfig {
    pub app_id: Option<String>,
    pub app_secret: Option<String>,
    pub stable_access_token_endpoint: String,
    pub virtual_payment_query_order_endpoint: String,
    pub virtual_payment_notify_provide_goods_endpoint: String,
}

#[derive(Clone, Debug)]
pub struct WechatClient {
    client: Client,
    config: WechatConfig,
    access_token_cache: Arc<Mutex<Option<WechatAccessTokenCacheEntry>>>,
}

#[derive(Clone, Debug)]
struct WechatAccessTokenCacheEntry {
    access_token: String,
    expires_at: Instant,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WechatVirtualPaymentQueryOrderRequest {
    pub openid: String,
    pub order_id: String,
    pub env: u8,
    pub app_key: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WechatVirtualPaymentOrder {
    pub order_id: String,
    pub status: i64,
    pub order_fee: u64,
    pub order_type: i64,
    pub paid_time: Option<i64>,
    pub wx_order_id: Option<String>,
    pub wxpay_order_id: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WechatVirtualPaymentNotifyProvideGoodsRequest {
    pub order_id: String,
    pub env: u8,
}

#[derive(Debug, PartialEq, Eq)]
pub enum WechatError {
    InvalidConfig(String),
    RequestFailed(String),
    DeserializeFailed(String),
    Upstream(String),
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WechatErrorKind {
    InvalidConfig,
    RequestFailed,
    DeserializeFailed,
    Upstream,
}

#[derive(Debug, Deserialize)]
struct WechatStableAccessTokenResponse {
    access_token: Option<String>,
    expires_in: Option<u64>,
    errcode: Option<i64>,
    errmsg: Option<String>,
}

#[derive(Debug, Serialize)]
struct WechatVirtualPaymentQueryOrderBody<'a> {
    openid: &'a str,
    env: u8,
    order_id: &'a str,
}

#[derive(Debug, Serialize)]
struct WechatVirtualPaymentNotifyProvideGoodsBody<'a> {
    order_id: &'a str,
    env: u8,
}

#[derive(Debug, Deserialize)]
struct WechatVirtualPaymentNotifyProvideGoodsResponse {
    #[serde(default)]
    errcode: i64,
    errmsg: Option<String>,
}

#[derive(Debug, Deserialize)]
struct WechatVirtualPaymentQueryOrderResponse {
    errcode: i64,
    errmsg: Option<String>,
    order: Option<WechatVirtualPaymentQueryOrderPayload>,
}

#[derive(Debug, Deserialize)]
struct WechatVirtualPaymentQueryOrderPayload {
    order_id: String,
    status: i64,
    order_fee: u64,
    order_type: i64,
    #[serde(default)]
    paid_time: Option<i64>,
    #[serde(default)]
    wx_order_id: Option<String>,
    #[serde(default)]
    wxpay_order_id: Option<String>,
}

impl WechatClient {
    pub fn new(config: WechatConfig) -> Self {
        Self {
            client: Client::new(),
            config,
            access_token_cache: Arc::new(Mutex::new(None)),
        }
    }

    pub async fn query_virtual_payment_order(
        &self,
        request: WechatVirtualPaymentQueryOrderRequest,
    ) -> Result<WechatVirtualPaymentOrder, WechatError> {
        let app_id = self
            .config
            .app_id
            .as_deref()
            .and_then(non_empty)
            .ok_or_else(|| WechatError::InvalidConfig("微信小程序 AppID 未配置".to_string()))?;
        let app_secret = self
            .config
            .app_secret
            .as_deref()
            .and_then(non_empty)
            .ok_or_else(|| WechatError::InvalidConfig("微信小程序 AppSecret 未配置".to_string()))?;
        let openid = non_empty(&request.openid)
            .ok_or_else(|| WechatError::InvalidConfig("微信虚拟支付查单缺少 openid".to_string()))?;
        let order_id = non_empty(&request.order_id).ok_or_else(|| {
            WechatError::InvalidConfig("微信虚拟支付查单缺少 order_id".to_string())
        })?;
        let app_key = non_empty(&request.app_key)
            .ok_or_else(|| WechatError::InvalidConfig("微信虚拟支付查单缺少 AppKey".to_string()))?;
        if request.env > 1 {
            return Err(WechatError::InvalidConfig(
                "微信虚拟支付查单 env 只允许 0 或 1".to_string(),
            ));
        }

        let body = serde_json::to_string(&WechatVirtualPaymentQueryOrderBody {
            openid,
            env: request.env,
            order_id,
        })
        .map_err(|error| {
            WechatError::DeserializeFailed(format!("微信虚拟支付查单请求序列化失败：{error}"))
        })?;
        let pay_sig = calc_virtual_payment_pay_signature(
            app_key,
            WECHAT_VIRTUAL_PAYMENT_QUERY_ORDER_URI,
            &body,
        );
        let mut access_token = self.request_access_token(app_id, app_secret, false).await?;
        let mut response = self
            .query_virtual_payment_order_with_token(&body, &pay_sig, &access_token)
            .await?;
        if is_access_token_error(response.errcode) {
            access_token = self
                .refresh_access_token_after_failure(app_id, app_secret, &access_token)
                .await?;
            response = self
                .query_virtual_payment_order_with_token(&body, &pay_sig, &access_token)
                .await?;
        }
        if response.errcode != 0 {
            return Err(WechatError::Upstream(format!(
                "微信虚拟支付查单返回错误：{}",
                response
                    .errmsg
                    .filter(|message| !message.trim().is_empty())
                    .unwrap_or_else(|| format!("errcode={}", response.errcode))
            )));
        }
        let order = response
            .order
            .ok_or_else(|| WechatError::Upstream("微信虚拟支付查单响应缺少 order".to_string()))?;
        if order.order_id != order_id {
            return Err(WechatError::Upstream(
                "微信虚拟支付查单返回的订单号与请求不一致".to_string(),
            ));
        }

        Ok(WechatVirtualPaymentOrder {
            order_id: order.order_id,
            status: order.status,
            order_fee: order.order_fee,
            order_type: order.order_type,
            paid_time: order.paid_time.filter(|value| *value > 0),
            wx_order_id: non_empty_owned_option(order.wx_order_id),
            wxpay_order_id: non_empty_owned_option(order.wxpay_order_id),
        })
    }

    pub async fn notify_virtual_payment_goods_delivered(
        &self,
        request: WechatVirtualPaymentNotifyProvideGoodsRequest,
    ) -> Result<(), WechatError> {
        let app_id = self
            .config
            .app_id
            .as_deref()
            .and_then(non_empty)
            .ok_or_else(|| WechatError::InvalidConfig("微信小程序 AppID 未配置".to_string()))?;
        let app_secret = self
            .config
            .app_secret
            .as_deref()
            .and_then(non_empty)
            .ok_or_else(|| WechatError::InvalidConfig("微信小程序 AppSecret 未配置".to_string()))?;
        let order_id = non_empty(&request.order_id).ok_or_else(|| {
            WechatError::InvalidConfig("微信虚拟支付发货确认缺少 order_id".to_string())
        })?;
        if request.env > 1 {
            return Err(WechatError::InvalidConfig(
                "微信虚拟支付发货确认 env 只允许 0 或 1".to_string(),
            ));
        }
        let body = serde_json::to_string(&WechatVirtualPaymentNotifyProvideGoodsBody {
            order_id,
            env: request.env,
        })
        .map_err(|error| {
            WechatError::DeserializeFailed(format!("微信虚拟支付发货确认请求序列化失败：{error}"))
        })?;

        let mut access_token = self.request_access_token(app_id, app_secret, false).await?;
        let mut response = self
            .notify_virtual_payment_goods_delivered_with_token(&body, &access_token)
            .await?;
        if is_access_token_error(response.errcode) {
            access_token = self
                .refresh_access_token_after_failure(app_id, app_secret, &access_token)
                .await?;
            response = self
                .notify_virtual_payment_goods_delivered_with_token(&body, &access_token)
                .await?;
        }
        if response.errcode != 0 {
            return Err(WechatError::Upstream(format!(
                "微信虚拟支付发货确认返回错误：{}",
                response
                    .errmsg
                    .filter(|message| !message.trim().is_empty())
                    .unwrap_or_else(|| format!("errcode={}", response.errcode))
            )));
        }
        Ok(())
    }

    async fn query_virtual_payment_order_with_token(
        &self,
        body: &str,
        pay_sig: &str,
        access_token: &str,
    ) -> Result<WechatVirtualPaymentQueryOrderResponse, WechatError> {
        let mut url =
            Url::parse(&self.config.virtual_payment_query_order_endpoint).map_err(|error| {
                WechatError::InvalidConfig(format!("微信虚拟支付查单地址非法：{error}"))
            })?;
        url.query_pairs_mut()
            .append_pair("access_token", access_token)
            .append_pair("pay_sig", pay_sig);
        let response = self
            .client
            .post(url.as_str())
            .header(reqwest::header::CONTENT_TYPE, "application/json")
            .body(body.to_string())
            .send()
            .await
            .map_err(|error| {
                warn!(error = %error, "微信虚拟支付查单请求失败");
                WechatError::RequestFailed("微信虚拟支付查单请求失败".to_string())
            })?;
        let status = response.status();
        let response_text = response.text().await.map_err(|error| {
            warn!(error = %error, "微信虚拟支付查单响应读取失败");
            WechatError::DeserializeFailed("微信虚拟支付查单响应读取失败".to_string())
        })?;
        if !status.is_success() {
            return Err(WechatError::Upstream(format!(
                "微信虚拟支付查单失败：HTTP {status}"
            )));
        }
        serde_json::from_str::<WechatVirtualPaymentQueryOrderResponse>(&response_text).map_err(
            |error| {
                warn!(error = %error, "微信虚拟支付查单响应解析失败");
                WechatError::DeserializeFailed("微信虚拟支付查单响应非法".to_string())
            },
        )
    }

    async fn notify_virtual_payment_goods_delivered_with_token(
        &self,
        body: &str,
        access_token: &str,
    ) -> Result<WechatVirtualPaymentNotifyProvideGoodsResponse, WechatError> {
        let mut url = Url::parse(&self.config.virtual_payment_notify_provide_goods_endpoint)
            .map_err(|error| {
                WechatError::InvalidConfig(format!("微信虚拟支付发货确认地址非法：{error}"))
            })?;
        url.query_pairs_mut()
            .append_pair("access_token", access_token);
        let response = self
            .client
            .post(url.as_str())
            .header(reqwest::header::CONTENT_TYPE, "application/json")
            .body(body.to_string())
            .send()
            .await
            .map_err(|error| {
                warn!(error = %error, "微信虚拟支付发货确认请求失败");
                WechatError::RequestFailed("微信虚拟支付发货确认请求失败".to_string())
            })?;
        let status = response.status();
        let response_text = response.text().await.map_err(|error| {
            warn!(error = %error, "微信虚拟支付发货确认响应读取失败");
            WechatError::DeserializeFailed("微信虚拟支付发货确认响应读取失败".to_string())
        })?;
        if !status.is_success() {
            return Err(WechatError::Upstream(format!(
                "微信虚拟支付发货确认失败：HTTP {status}"
            )));
        }
        if response_text.trim().is_empty() {
            return Ok(WechatVirtualPaymentNotifyProvideGoodsResponse {
                errcode: 0,
                errmsg: None,
            });
        }
        serde_json::from_str::<WechatVirtualPaymentNotifyProvideGoodsResponse>(&response_text)
            .map_err(|error| {
                warn!(error = %error, "微信虚拟支付发货确认响应解析失败");
                WechatError::DeserializeFailed("微信虚拟支付发货确认响应非法".to_string())
            })
    }

    async fn refresh_access_token_after_failure(
        &self,
        app_id: &str,
        app_secret: &str,
        failed_access_token: &str,
    ) -> Result<String, WechatError> {
        let force_refresh = {
            let mut cache = self.access_token_cache.lock().await;
            if let Some(entry) = cache.as_ref()
                && entry.access_token != failed_access_token
                && Instant::now() < entry.expires_at
            {
                return Ok(entry.access_token.clone());
            }
            if cache
                .as_ref()
                .is_some_and(|entry| entry.access_token == failed_access_token)
            {
                *cache = None;
            }
            true
        };
        self.request_access_token(app_id, app_secret, force_refresh)
            .await
    }

    async fn request_access_token(
        &self,
        app_id: &str,
        app_secret: &str,
        force_refresh: bool,
    ) -> Result<String, WechatError> {
        let mut cache = self.access_token_cache.lock().await;
        if !force_refresh
            && let Some(entry) = cache.as_ref()
            && Instant::now() < entry.expires_at
        {
            return Ok(entry.access_token.clone());
        }

        let url = Url::parse(&self.config.stable_access_token_endpoint).map_err(|error| {
            WechatError::InvalidConfig(format!("微信 stable_token 地址非法：{error}"))
        })?;
        let payload = self
            .client
            .post(url.as_str())
            .json(&json!({
                "grant_type": "client_credential",
                "appid": app_id,
                "secret": app_secret,
                "force_refresh": force_refresh,
            }))
            .send()
            .await
            .map_err(|error| {
                warn!(error = %error, "微信 stable_token 请求失败");
                WechatError::RequestFailed("微信 stable_token 请求失败".to_string())
            })?
            .json::<WechatStableAccessTokenResponse>()
            .await
            .map_err(|error| {
                warn!(error = %error, "微信 stable_token 响应解析失败");
                WechatError::DeserializeFailed("微信 stable_token 响应非法".to_string())
            })?;

        if let Some(errcode) = payload.errcode.filter(|value| *value != 0) {
            return Err(WechatError::Upstream(format!(
                "微信 stable_token 返回错误：{}",
                payload
                    .errmsg
                    .unwrap_or_else(|| format!("errcode={errcode}"))
            )));
        }

        let expires_in = payload.expires_in.unwrap_or(7_200);
        let access_token = payload
            .access_token
            .and_then(|value| non_empty_owned(value))
            .ok_or_else(|| {
                WechatError::Upstream("微信 stable_token 缺少 access_token".to_string())
            })?;
        let cache_lifetime = Duration::from_secs(expires_in)
            .saturating_sub(WECHAT_ACCESS_TOKEN_REFRESH_SAFETY_MARGIN);
        if !cache_lifetime.is_zero()
            && let Some(expires_at) = Instant::now().checked_add(cache_lifetime)
        {
            *cache = Some(WechatAccessTokenCacheEntry {
                access_token: access_token.clone(),
                expires_at,
            });
        }
        Ok(access_token)
    }
}

impl WechatError {
    pub fn kind(&self) -> WechatErrorKind {
        match self {
            Self::InvalidConfig(_) => WechatErrorKind::InvalidConfig,
            Self::RequestFailed(_) => WechatErrorKind::RequestFailed,
            Self::DeserializeFailed(_) => WechatErrorKind::DeserializeFailed,
            Self::Upstream(_) => WechatErrorKind::Upstream,
        }
    }
}

impl fmt::Display for WechatError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidConfig(message)
            | Self::RequestFailed(message)
            | Self::DeserializeFailed(message)
            | Self::Upstream(message) => f.write_str(message),
        }
    }
}

impl Error for WechatError {}

fn non_empty(value: &str) -> Option<&str> {
    let trimmed = value.trim();
    if trimmed.is_empty() {
        None
    } else {
        Some(trimmed)
    }
}

fn non_empty_owned(value: String) -> Option<String> {
    if value.trim().is_empty() {
        None
    } else {
        Some(value)
    }
}

fn non_empty_owned_option(value: Option<String>) -> Option<String> {
    value.and_then(non_empty_owned)
}

fn is_access_token_error(errcode: i64) -> bool {
    WECHAT_ACCESS_TOKEN_ERROR_CODES.contains(&errcode)
}

fn calc_virtual_payment_pay_signature(app_key: &str, uri: &str, body: &str) -> String {
    let key = hmac::Key::new(hmac::HMAC_SHA256, app_key.as_bytes());
    hex::encode(hmac::sign(&key, format!("{uri}&{body}").as_bytes()).as_ref())
}

#[cfg(test)]
mod tests {
    use std::{
        io::{Read, Write},
        net::TcpListener,
        sync::mpsc,
        thread,
    };

    use super::{
        WECHAT_VIRTUAL_PAYMENT_QUERY_ORDER_URI, WechatClient, WechatConfig,
        WechatVirtualPaymentNotifyProvideGoodsRequest, WechatVirtualPaymentQueryOrderBody,
        WechatVirtualPaymentQueryOrderRequest, calc_virtual_payment_pay_signature,
    };

    #[test]
    fn virtual_payment_pay_signature_matches_official_example() {
        let body = r#"{"openid": "xxx", "user_ip": "127.0.0.1", "env": 0}"#;

        assert_eq!(
            calc_virtual_payment_pay_signature("12345", "/xpay/query_user_balance", body),
            "c37809f27c6d7fd1837ad2500a04512b66b34fd793a39a385fade56dca89a4b5"
        );
    }

    #[test]
    fn virtual_payment_query_order_signs_the_exact_serialized_body() {
        let body = serde_json::to_string(&WechatVirtualPaymentQueryOrderBody {
            openid: "openid-001",
            env: 0,
            order_id: "order-001",
        })
        .expect("query body should serialize");

        assert_eq!(
            body,
            r#"{"openid":"openid-001","env":0,"order_id":"order-001"}"#
        );
        assert_eq!(
            calc_virtual_payment_pay_signature(
                "app-key-001",
                WECHAT_VIRTUAL_PAYMENT_QUERY_ORDER_URI,
                &body,
            ),
            "ca0540a55865df7b4dbd7400d4d7a6551fde4f66fff9f6290f40f23e32d33db8"
        );
    }

    #[tokio::test]
    async fn virtual_payment_query_order_uses_the_configured_mock_endpoint() {
        let listener = TcpListener::bind("127.0.0.1:0").expect("mock listener should bind");
        let address = listener.local_addr().expect("mock address should resolve");
        let (request_sender, request_receiver) = mpsc::channel();
        let server = thread::spawn(move || {
            for response_body in [
                r#"{"access_token":"access-token-001","expires_in":7200}"#,
                r#"{"errcode":0,"errmsg":"ok","order":{"order_id":"order-001","status":2,"order_fee":600,"order_type":0,"paid_time":1777111300,"wx_order_id":"wx-order-001","wxpay_order_id":"wxpay-order-001"}}"#,
            ] {
                let (mut stream, _) = listener.accept().expect("mock request should connect");
                let request = read_http_request(&mut stream);
                request_sender
                    .send(request)
                    .expect("mock request should be recorded");
                let response = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    response_body.len(),
                    response_body
                );
                stream
                    .write_all(response.as_bytes())
                    .expect("mock response should write");
            }
        });
        let client = WechatClient::new(WechatConfig {
            app_id: Some("wx-app-001".to_string()),
            app_secret: Some("app-secret-001".to_string()),
            stable_access_token_endpoint: format!("http://{address}/stable-token"),
            virtual_payment_query_order_endpoint: format!("http://{address}/mock/query-order"),
            virtual_payment_notify_provide_goods_endpoint: format!(
                "http://{address}/mock/notify-provide-goods"
            ),
        });

        let order = client
            .query_virtual_payment_order(WechatVirtualPaymentQueryOrderRequest {
                openid: "openid-001".to_string(),
                order_id: "order-001".to_string(),
                env: 0,
                app_key: "app-key-001".to_string(),
            })
            .await
            .expect("mock query should succeed");

        assert_eq!(order.status, 2);
        assert_eq!(order.order_fee, 600);
        let token_request = request_receiver
            .recv()
            .expect("stable token request should be recorded");
        let query_request = request_receiver
            .recv()
            .expect("query request should be recorded");
        assert!(token_request.starts_with("POST /stable-token HTTP/1.1"));
        assert!(query_request.starts_with("POST /mock/query-order?access_token="));
        assert!(query_request.contains("&pay_sig="));
        assert!(
            query_request.ends_with(r#"{"openid":"openid-001","env":0,"order_id":"order-001"}"#)
        );
        server.join().expect("mock server should finish");
    }

    #[tokio::test]
    async fn virtual_payment_goods_delivery_accepts_an_empty_success_response() {
        let listener = TcpListener::bind("127.0.0.1:0").expect("mock listener should bind");
        let address = listener.local_addr().expect("mock address should resolve");
        let (request_sender, request_receiver) = mpsc::channel();
        let server = thread::spawn(move || {
            for response_body in [
                r#"{"access_token":"access-token-001","expires_in":7200}"#,
                "",
            ] {
                let (mut stream, _) = listener.accept().expect("mock request should connect");
                let request = read_http_request(&mut stream);
                request_sender
                    .send(request)
                    .expect("mock request should be recorded");
                let response = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    response_body.len(),
                    response_body
                );
                stream
                    .write_all(response.as_bytes())
                    .expect("mock response should write");
            }
        });
        let client = test_wechat_client(address);

        client
            .notify_virtual_payment_goods_delivered(WechatVirtualPaymentNotifyProvideGoodsRequest {
                order_id: "order-001".to_string(),
                env: 0,
            })
            .await
            .expect("empty 200 response should confirm delivery");

        let _token_request = request_receiver
            .recv()
            .expect("stable token request should be recorded");
        let notify_request = request_receiver
            .recv()
            .expect("provide goods request should be recorded");
        assert!(
            notify_request.starts_with(
                "POST /mock/notify-provide-goods?access_token=access-token-001 HTTP/1.1"
            )
        );
        assert!(notify_request.ends_with(r#"{"order_id":"order-001","env":0}"#));
        server.join().expect("mock server should finish");
    }

    #[tokio::test]
    async fn virtual_payment_query_refreshes_an_invalid_cached_access_token_once() {
        let listener = TcpListener::bind("127.0.0.1:0").expect("mock listener should bind");
        let address = listener.local_addr().expect("mock address should resolve");
        let (request_sender, request_receiver) = mpsc::channel();
        let server = thread::spawn(move || {
            for response_body in [
                r#"{"access_token":"stale-token","expires_in":7200}"#,
                r#"{"errcode":40014,"errmsg":"invalid access token"}"#,
                r#"{"access_token":"fresh-token","expires_in":7200}"#,
                r#"{"errcode":0,"errmsg":"ok","order":{"order_id":"order-001","status":2,"order_fee":600,"order_type":0,"paid_time":1777111300}}"#,
            ] {
                let (mut stream, _) = listener.accept().expect("mock request should connect");
                let request = read_http_request(&mut stream);
                request_sender
                    .send(request)
                    .expect("mock request should be recorded");
                let response = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    response_body.len(),
                    response_body
                );
                stream
                    .write_all(response.as_bytes())
                    .expect("mock response should write");
            }
        });
        let client = test_wechat_client(address);

        let order = client
            .query_virtual_payment_order(WechatVirtualPaymentQueryOrderRequest {
                openid: "openid-001".to_string(),
                order_id: "order-001".to_string(),
                env: 0,
                app_key: "app-key-001".to_string(),
            })
            .await
            .expect("query should retry with a refreshed token");
        assert_eq!(order.status, 2);

        let first_token_request = request_receiver.recv().expect("first token request");
        let first_query_request = request_receiver.recv().expect("first query request");
        let second_token_request = request_receiver.recv().expect("second token request");
        let second_query_request = request_receiver.recv().expect("second query request");
        assert!(first_token_request.contains(r#""force_refresh":false"#));
        assert!(first_query_request.contains("access_token=stale-token"));
        assert!(second_token_request.contains(r#""force_refresh":true"#));
        assert!(second_query_request.contains("access_token=fresh-token"));
        server.join().expect("mock server should finish");
    }

    fn test_wechat_client(address: std::net::SocketAddr) -> WechatClient {
        WechatClient::new(WechatConfig {
            app_id: Some("wx-app-001".to_string()),
            app_secret: Some("app-secret-001".to_string()),
            stable_access_token_endpoint: format!("http://{address}/stable-token"),
            virtual_payment_query_order_endpoint: format!("http://{address}/mock/query-order"),
            virtual_payment_notify_provide_goods_endpoint: format!(
                "http://{address}/mock/notify-provide-goods"
            ),
        })
    }

    fn read_http_request(stream: &mut std::net::TcpStream) -> String {
        let mut bytes = Vec::new();
        let mut buffer = [0_u8; 4096];
        loop {
            let count = stream.read(&mut buffer).expect("mock request should read");
            if count == 0 {
                break;
            }
            bytes.extend_from_slice(&buffer[..count]);
            if let Some(header_end) = bytes.windows(4).position(|window| window == b"\r\n\r\n") {
                let headers = String::from_utf8_lossy(&bytes[..header_end + 4]);
                let content_length = headers
                    .lines()
                    .find_map(|line| {
                        line.strip_prefix("content-length:")
                            .or_else(|| line.strip_prefix("Content-Length:"))
                    })
                    .and_then(|value| value.trim().parse::<usize>().ok())
                    .unwrap_or(0);
                if bytes.len() >= header_end + 4 + content_length {
                    break;
                }
            }
        }
        String::from_utf8(bytes).expect("mock request should be UTF-8")
    }
}