robit_qq/

platform.rs

//! QQ Official Bot platform adapter.
//!
//! Implements [`robit_chatbot::PlatformAdapter`] for the QQ Official Bot API:
//!
//! - **Access token**: obtained via the OAuth2 `getAppAccessToken` endpoint
//!   (app_id + app_secret), refreshed before expiry, and used for both the
//!   WebSocket Identify and HTTP message sends.
//! - **WebSocket gateway**: connects, sends Identify (`op=2`), then runs a
//!   heartbeat task and a dispatch task. Dispatch converts
//!   `C2C_MESSAGE_CREATE` / `GROUP_AT_MESSAGE_CREATE` events into
//!   [`PlatformEvent::Message`].
//! - **HTTP send**: POSTs text to the group or user messages endpoint.
//!
//! `chat_id` encoding: `"group:{group_openid}"` for group chats,
//! `"private:{user_openid}"` for C2C chats.

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

use async_trait::async_trait;
use futures_util::{SinkExt, StreamExt};
use robit_agent::error::{AgentError, Result};
use robit_ai::config::RobitConfig;
use robit_chatbot::adapter::{
    ChatMessage, ChatType, MediaAttachment, PlatformAdapter, PlatformCaps, PlatformEvent,
    SendResult, SenderInfo, UploadResult,
};
use tokio::sync::{mpsc, Mutex, RwLock};
use tokio_util::sync::CancellationToken;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, info, warn};

use crate::protocol::{
    event_type, AccessTokenRequest, AccessTokenResponse, GatewayPayload, HelloData, MediaFileInfo,
    MessageEvent, SendMessageRequest, SendMessageResponse, op,
};

/// QQ Bot configuration parsed from `[channels.qq_bot]`.
#[derive(Debug, Clone)]
pub struct QqConfig {
    pub app_id: String,
    pub app_secret: String,
    pub sandbox: bool,
}

impl QqConfig {
    /// Extract QQ Bot config from the loaded `RobitConfig`.
    pub fn from_config(config: &RobitConfig) -> std::result::Result<Self, String> {
        let qq = config
            .channels
            .as_ref()
            .and_then(|c| c.qq_bot.as_ref())
            .ok_or_else(|| {
                "QQ Bot config not found. Add [channels.qq_bot] section to config.toml".to_string()
            })?;
        Ok(Self {
            app_id: qq.app_id.clone(),
            app_secret: qq.app_secret.clone(),
            sandbox: false,
        })
    }

    /// WebSocket gateway URL.
    pub fn gateway_url(&self) -> &str {
        if self.sandbox {
            "wss://sandbox.api.sgroup.qq.com/websockets"
        } else {
            "wss://api.sgroup.qq.com/websockets"
        }
    }

    /// HTTP API base URL.
    pub fn api_base_url(&self) -> &str {
        if self.sandbox {
            "https://sandbox.api.sgroup.qq.com"
        } else {
            "https://api.sgroup.qq.com"
        }
    }

    /// App access token endpoint.
    pub fn access_token_url(&self) -> &str {
        "https://bots.qq.com/app/getAppAccessToken"
    }
}

/// A cached access token with its expiry.
struct CachedToken {
    token: String,
    /// Instant at which the token expires.
    expires_at: Instant,
}

/// QQ Official Bot platform adapter.
pub struct QqPlatformAdapter {
    config: QqConfig,
    /// HTTP client for sending messages and fetching access tokens.
    http: reqwest::Client,
    /// Cached app access token (refreshed as needed).
    access_token: RwLock<Option<CachedToken>>,
    /// Last sequence number received (for heartbeats / resume).
    last_seq: Mutex<Option<u64>>,
    /// Session ID from the Ready event (for resume).
    session_id: Mutex<Option<String>>,
    /// Inbound event channel: dispatch/heartbeat tasks push, recv_event pops.
    event_tx: mpsc::Sender<PlatformEvent>,
    event_rx: Mutex<mpsc::Receiver<PlatformEvent>>,
    /// Outbound WebSocket writes (shared between send_message-via-WS and the
    /// heartbeat task). Currently send_message uses HTTP, so this is owned by
    /// the dispatch/heartbeat tasks.
    ws_tx: Mutex<Option<futures_util::stream::SplitSink<WebSocket, Message>>>,
    /// Platform capabilities (kept for diagnostics; capabilities() is static).
    #[allow(dead_code)]
    caps: PlatformCaps,
    /// Heartbeat interval (from the Hello event).
    heartbeat_interval: RwLock<Duration>,
    /// Tracks the last received message ID per chat, so passive replies can
    /// reference it (QQ requires `msg_id` within 5 min of the original).
    last_inbound_msg_id: Mutex<Option<(String, String)>>, // (chat_id, msg_id)
    /// Monotonic counter for `msg_seq` per reply.
    msg_seq: Mutex<u32>,
}

type WebSocket =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

impl QqPlatformAdapter {
    /// Connect to the QQ gateway and start the heartbeat + dispatch tasks.
    ///
    /// This is the constructor used in place of `PlatformAdapter::connect` when
    /// we already own the config (avoids the `Self::Config` indirection).
    /// `shutdown` is notified on Ctrl+C to let background tasks exit gracefully.
    ///
    /// The first connection is established synchronously so bad config (wrong
    /// app_id/secret, unreachable gateway) surfaces immediately. A background
    /// supervisor then owns the connection lifecycle and transparently
    /// reconnects on gateway rotations (roughly every ~30 min) without
    /// dropping the adapter or the agents held by the `ChatbotManager`.
    pub async fn connect(config: QqConfig, shutdown: Arc<tokio::sync::Notify>) -> Result<Arc<Self>> {
        let http = reqwest::Client::new();
        let caps = PlatformCaps::qq();
        let (event_tx, event_rx) = mpsc::channel::<PlatformEvent>(256);

        let adapter = Arc::new(Self {
            config: config.clone(),
            http,
            access_token: RwLock::new(None),
            last_seq: Mutex::new(None),
            session_id: Mutex::new(None),
            event_tx,
            event_rx: Mutex::new(event_rx),
            ws_tx: Mutex::new(None),
            caps,
            heartbeat_interval: RwLock::new(Duration::from_secs(41)),
            last_inbound_msg_id: Mutex::new(None),
            msg_seq: Mutex::new(0),
        });

        let conn_cancel = CancellationToken::new();
        let tasks = adapter
            .establish_connection(shutdown.clone(), conn_cancel.clone())
            .await?;

        // Supervisor owns all subsequent reconnects. As long as it can
        // re-establish the WebSocket, the adapter - and the agents held by the
        // ChatbotManager - survive across QQ gateway rotations, which
        // previously killed the whole process along with every in-flight async
        // task (e.g. image generation).
        let supervisor = Arc::clone(&adapter);
        tokio::spawn(async move {
            supervisor_loop(supervisor, tasks, conn_cancel, shutdown).await;
        });

        Ok(adapter)
    }

    /// Open the WebSocket, complete the Hello → Identify handshake, and spawn
    /// the heartbeat + dispatch background tasks. `conn_cancel` lets the
    /// supervisor stop both tasks when it tears the connection down for a
    /// reconnect; `shutdown` lets them exit on Ctrl+C. Neither task emits
    /// `Disconnected` - the supervisor owns reconnect, so the manager/agents
    /// stay alive across gateway rotations.
    async fn establish_connection(
        self: &Arc<Self>,
        shutdown: Arc<tokio::sync::Notify>,
        conn_cancel: CancellationToken,
    ) -> Result<ConnectionTasks> {
        info!("Connecting to QQ gateway: {}", self.config.gateway_url());
        let (ws_stream, _response) = tokio_tungstenite::connect_async(self.config.gateway_url())
            .await
            .map_err(|e| AgentError::InternalError(format!("WebSocket connect failed: {}", e)))?;

        let (mut write, mut read) = ws_stream.split();
        write
            .send(Message::Ping(bytes::Bytes::new()))
            .await
            .map_err(|e| AgentError::InternalError(format!("WS ping failed: {}", e)))?;

        // 1. Wait for Hello (op=10) to learn the heartbeat interval.
        let heartbeat_interval = loop {
            let msg = read
                .next()
                .await
                .ok_or_else(|| AgentError::InternalError("WebSocket closed before Hello".into()))?
                .map_err(|e| AgentError::InternalError(format!("WS read error: {}", e)))?;
            if let Message::Text(text) = msg {
                let payload: GatewayPayload =
                    serde_json::from_str(&text).map_err(|e| {
                        AgentError::InternalError(format!("Invalid Hello JSON: {}", e))
                    })?;
                if payload.op == op::HELLO {
                    let hello: HelloData = serde_json::from_value(
                        payload.d.ok_or_else(|| AgentError::InternalError("Hello missing d".into()))?,
                    )
                    .map_err(|e| AgentError::InternalError(format!("Invalid Hello data: {}", e)))?;
                    break Duration::from_millis(hello.heartbeat_interval);
                }
            }
        };
        *self.heartbeat_interval.write().await = heartbeat_interval;
        info!("QQ heartbeat interval: {:?}", heartbeat_interval);

        // 2. Fetch an app access token and send Identify (op=2).
        let access_token = self.fetch_access_token().await?;
        let identify =
            GatewayPayload::identify(&access_token, INTENT_C2C | INTENT_GROUP_AT_MESSAGE);
        write
            .send(Message::Text(serde_json::to_string(&identify).unwrap().into()))
            .await
            .map_err(|e| AgentError::InternalError(format!("Identify send failed: {}", e)))?;

        // 3. Store the write half and spawn the heartbeat + dispatch tasks.
        *self.ws_tx.lock().await = Some(write);

        let heartbeat = spawn_heartbeat(Arc::clone(self), conn_cancel.clone(), shutdown.clone());
        let dispatch = spawn_dispatch(Arc::clone(self), read, conn_cancel, shutdown);

        Ok(ConnectionTasks { dispatch, heartbeat })
    }

    /// Fetch (and cache) an app access token, returning a fresh one.
    async fn fetch_access_token(&self) -> Result<String> {
        // Return cached if still valid (with a 60s safety margin).
        {
            let cache = self.access_token.read().await;
            if let Some(cached) = cache.as_ref() {
                if cached.expires_at.duration_since(Instant::now())
                    > Duration::from_secs(60)
                {
                    return Ok(cached.token.clone());
                }
            }
        }

        let req = AccessTokenRequest {
            app_id: self.config.app_id.clone(),
            client_secret: self.config.app_secret.clone(),
        };

        let response = self
            .http
            .post(self.config.access_token_url())
            .json(&req)
            .send()
            .await
            .map_err(|e| AgentError::InternalError(format!("Access token request failed: {}", e)))?;

        let status = response.status();
        let text = response.text().await.unwrap_or_default();

        if !status.is_success() {
            return Err(AgentError::InternalError(format!("Access token request failed ({}): {}", status, text)));
        }

        let resp: AccessTokenResponse = serde_json::from_str(&text)
            .map_err(|e| AgentError::InternalError(format!("Access token parse failed: {}", e)))?;

        let token = resp.access_token.clone();
        let expires_in = resp.expires_in.max(60);
        let cached = CachedToken {
            token: token.clone(),
            expires_at: Instant::now() + Duration::from_secs(expires_in),
        };
        *self.access_token.write().await = Some(cached);
        debug!("Fetched QQ access token (expires in {}s)", expires_in);
        Ok(token)
    }

    /// Build the Authorization header value (`QQBot {token}`).
    async fn auth_header(&self) -> Result<String> {
        let token = self.fetch_access_token().await?;
        Ok(format!("QQBot {}", token))
    }

    /// Record the inbound message ID for a chat (so a later reply can reference it).
    fn record_inbound(&self, chat_id: &str, msg_id: &str) {
        if let Ok(mut guard) = self.last_inbound_msg_id.try_lock() {
            *guard = Some((chat_id.to_string(), msg_id.to_string()));
        }
    }

    /// The inbound message ID to reference for a reply to `chat_id`, if any.
    async fn reply_msg_id(&self, chat_id: &str) -> Option<String> {
        let guard = self.last_inbound_msg_id.lock().await;
        guard
            .as_ref()
            .filter(|(cid, _)| cid == chat_id)
            .map(|(_, id)| id.clone())
    }

    async fn next_msg_seq(&self) -> u32 {
        let mut seq = self.msg_seq.lock().await;
        *seq = seq.wrapping_add(1);
        *seq
    }
}

#[async_trait]
impl PlatformAdapter for QqPlatformAdapter {
    fn capabilities() -> PlatformCaps {
        PlatformCaps::qq()
    }

    async fn send_message(&self, chat_id: &str, text: &str) -> Result<SendResult> {
        let auth = self.auth_header().await?;
        let (endpoint, is_group) = resolve_send_endpoint(self.config.api_base_url(), chat_id)?;

        let msg_id = self.reply_msg_id(chat_id).await;
        let msg_seq = self.next_msg_seq().await;
        let body = SendMessageRequest {
            content: text.to_string(),
            msg_type: crate::protocol::msg_type::TEXT,
            msg_id,
            msg_seq: Some(msg_seq),
            media: None,
        };

        let resp = self
            .http
            .post(&endpoint)
            .header("Authorization", &auth)
            .json(&body)
            .send()
            .await
            .map_err(|e| AgentError::InternalError(format!("QQ send failed: {}", e)))?;

        let status = resp.status();
        if !status.is_success() {
            let text = resp.text().await.unwrap_or_default();
            warn!("QQ send {} failed ({}): {}", endpoint, status, text);
            return Err(AgentError::InternalError(format!(
                "QQ send failed ({})",
                status
            )));
        }

        let parsed: SendMessageResponse = resp
            .json()
            .await
            .map_err(|e| AgentError::InternalError(format!("QQ send response parse: {}", e)))?;

        let id = parsed
            .id
            .or(parsed.msg_id)
            .unwrap_or_else(|| format!("sent-{}", msg_seq));
        debug!("QQ message sent to {} (id={})", chat_id, id);
        let _ = is_group; // currently unused beyond endpoint selection
        Ok(SendResult { msg_id: id })
    }

    async fn edit_message(&self, _chat_id: &str, _msg_id: &str, _text: &str) -> Result<()> {
        // QQ's passive-reply model doesn't support editing a sent message in
        // place (each reply is a new message referencing a msg_id). Fall back
        // to a fresh send so edit-based streaming degrades gracefully.
        self.send_message(_chat_id, _text).await?;
        Ok(())
    }

    async fn upload_file(
        &self,
        chat_id: &str,
        file_path: &str,
        media_type: &str,
    ) -> Result<UploadResult> {
        let auth = self.auth_header().await?;
        let (endpoint, _is_group) =
            resolve_upload_endpoint(self.config.api_base_url(), chat_id)?;

        let file_type = match media_type {
            "image" => crate::protocol::file_type::IMAGE,
            "video" => crate::protocol::file_type::VIDEO,
            "voice" => crate::protocol::file_type::VOICE,
            _ => crate::protocol::file_type::FILE,
        };

        // Read file from disk and encode as base64.
        let file_data = tokio::fs::read(file_path)
            .await
            .map_err(|e| AgentError::InternalError(format!("Failed to read file {}: {}", file_path, e)))?;

        use base64::Engine;
        let file_data_b64 = base64::engine::general_purpose::STANDARD.encode(&file_data);

        // QQ upload API uses JSON body with base64-encoded file_data.
        // srv_send_msg = false → returns file_info for later use (recommended).
        let body = crate::protocol::UploadMediaRequest {
            file_type,
            url: None,
            file_data: Some(file_data_b64),
            srv_send_msg: false,
        };

        let resp = self
            .http
            .post(&endpoint)
            .header("Authorization", &auth)
            .json(&body)
            .send()
            .await
            .map_err(|e| AgentError::InternalError(format!("QQ upload failed: {}", e)))?;

        let status = resp.status();
        let resp_body = resp.text().await.unwrap_or_default();

        if !status.is_success() {
            warn!("QQ upload {} failed ({}): {}", endpoint, status, resp_body);
            return Err(AgentError::InternalError(format!(
                "QQ upload failed ({}): {}",
                status, resp_body
            )));
        }

        let parsed: crate::protocol::UploadMediaResponse = serde_json::from_str(&resp_body)
            .map_err(|e| {
                AgentError::InternalError(format!(
                    "QQ upload response parse failed: {} (body: {})",
                    e, resp_body
                ))
            })?;

        let file_info = parsed.file_info.ok_or_else(|| {
            AgentError::InternalError(format!(
                "QQ upload response missing file_info: {}",
                resp_body
            ))
        })?;

        let file_id = parsed
            .file_uuid
            .or(parsed.id)
            .unwrap_or_else(|| file_info.clone());

        debug!("QQ file uploaded: file_info={}, file_id={}", file_info, file_id);

        Ok(UploadResult {
            file_id,
            url: file_info,
        })
    }

    async fn send_media_message(
        &self,
        chat_id: &str,
        file_url: &str,
        file_name: &str,
        media_type: &str,
    ) -> Result<SendResult> {
        let auth = self.auth_header().await?;
        let (endpoint, _is_group) = resolve_send_endpoint(self.config.api_base_url(), chat_id)?;

        let msg_id = self.reply_msg_id(chat_id).await;
        let msg_seq = self.next_msg_seq().await;
        let body = SendMessageRequest {
            content: file_name.to_string(),
            msg_type: crate::protocol::msg_type::MEDIA,
            msg_id,
            msg_seq: Some(msg_seq),
            media: Some(MediaFileInfo {
                file_info: file_url.to_string(),
            }),
        };

        let resp = self
            .http
            .post(&endpoint)
            .header("Authorization", &auth)
            .json(&body)
            .send()
            .await
            .map_err(|e| AgentError::InternalError(format!("QQ media send failed: {}", e)))?;

        let status = resp.status();
        if !status.is_success() {
            let text = resp.text().await.unwrap_or_default();
            warn!("QQ media send {} failed ({}): {}", endpoint, status, text);
            return Err(AgentError::InternalError(format!(
                "QQ media send failed ({}): {}",
                status, text
            )));
        }

        let parsed: SendMessageResponse = resp
            .json()
            .await
            .map_err(|e| {
                AgentError::InternalError(format!("QQ media send response parse: {}", e))
            })?;

        let id = parsed
            .id
            .or(parsed.msg_id)
            .unwrap_or_else(|| format!("media-{}", msg_seq));
        debug!(
            "QQ media message sent to {} (id={}, type={})",
            chat_id, id, media_type
        );
        Ok(SendResult { msg_id: id })
    }

    async fn recv_event(&self) -> Result<PlatformEvent> {
        self.event_rx
            .lock()
            .await
            .recv()
            .await
            .ok_or_else(|| AgentError::InternalError("QQ event channel closed".into()))
    }
}

/// Intent bitmask for C2C + group @-messages.
const INTENT_C2C: u32 = crate::protocol::INTENT_C2C;
const INTENT_GROUP_AT_MESSAGE: u32 = crate::protocol::INTENT_GROUP_AT_MESSAGE;

/// Resolve the HTTP send endpoint for a chat_id.
///
/// `group:{openid}` → `/v2/groups/{openid}/messages`
/// `private:{openid}` → `/v2/users/{openid}/messages`
fn resolve_send_endpoint(base: &str, chat_id: &str) -> Result<(String, bool)> {
    if let Some(group_id) = chat_id.strip_prefix("group:") {
        return Ok((
            format!("{}/v2/groups/{}/messages", base, group_id),
            true,
        ));
    }
    if let Some(user_id) = chat_id.strip_prefix("private:") {
        return Ok((
            format!("{}/v2/users/{}/messages", base, user_id),
            false,
        ));
    }
    Err(AgentError::InternalError(format!(
        "Invalid chat_id '{}': expected 'group:{{id}}' or 'private:{{id}}'",
        chat_id
    )))
}

/// Resolve the HTTP upload endpoint for a chat_id.
///
/// `group:{openid}` → `/v2/groups/{openid}/files`
/// `private:{openid}` → `/v2/users/{openid}/files`
fn resolve_upload_endpoint(base: &str, chat_id: &str) -> Result<(String, bool)> {
    if let Some(group_id) = chat_id.strip_prefix("group:") {
        return Ok((
            format!("{}/v2/groups/{}/files", base, group_id),
            true,
        ));
    }
    if let Some(user_id) = chat_id.strip_prefix("private:") {
        return Ok((
            format!("{}/v2/users/{}/files", base, user_id),
            false,
        ));
    }
    Err(AgentError::InternalError(format!(
        "Invalid chat_id '{}': expected 'group:{{id}}' or 'private:{{id}}'",
        chat_id
    )))
}

/// Bookkeeping for the two background tasks of one WebSocket connection.
struct ConnectionTasks {
    dispatch: tokio::task::JoinHandle<()>,
    heartbeat: tokio::task::JoinHandle<()>,
}

/// Action returned by frame processing: keep going, or drop the connection
/// (the supervisor will reconnect).
#[derive(Clone, Copy)]
enum FrameAction {
    Continue,
    Reconnect,
}

/// Which connection task ended first in the supervisor's `select!`. The
/// `select!` consumes the JoinHandle of whichever task fires, so the
/// supervisor must await only the *other* task afterwards - re-polling the
/// consumed handle panics ("JoinHandle polled after completion").
#[derive(Clone, Copy)]
enum TaskEnded {
    Dispatch,
    Heartbeat,
}

/// Background reconnect supervisor: owns the connection lifecycle after the
/// initial [`QqPlatformAdapter::connect`]. When either the dispatch or
/// heartbeat task of the current connection ends (routine QQ gateway rotation
/// every ~30 min, network blip, send/read failure, server-requested
/// reconnect), it cancels the other task and re-establishes the WebSocket
/// with bounded backoff. The adapter - and the agents held by the
/// `ChatbotManager` - stay alive across reconnects; only `shutdown` (Ctrl+C)
/// stops the supervisor.
async fn supervisor_loop(
    adapter: Arc<QqPlatformAdapter>,
    mut tasks: ConnectionTasks,
    mut conn_cancel: CancellationToken,
    shutdown: Arc<tokio::sync::Notify>,
) {
    let mut establish_failures: u32 = 0;
    loop {
        // Wait for either connection task to end, or for shutdown. When one
        // task ends the connection is considered lost (the other is likely
        // already failing or about to). `select!` consumes the JoinHandle of
        // whichever task fires, so below we await only the *other* one:
        // re-polling the consumed handle panics ("JoinHandle polled after
        // completion"), which used to silently kill the supervisor (the panic
        // went to stderr, not the tracing log) and left QQ offline after every
        // ~30-min gateway rotation.
        let ended = tokio::select! {
            res = &mut tasks.dispatch => {
                match res {
                    Ok(()) => debug!("Dispatch task ended"),
                    Err(e) => warn!("Dispatch task panicked: {}", e),
                }
                TaskEnded::Dispatch
            }
            res = &mut tasks.heartbeat => {
                match res {
                    Ok(()) => debug!("Heartbeat task ended"),
                    Err(e) => warn!("Heartbeat task panicked: {}", e),
                }
                TaskEnded::Heartbeat
            }
            _ = shutdown.notified() => {
                debug!("Supervisor received shutdown signal");
                conn_cancel.cancel();
                // Neither handle was consumed (the shutdown branch fired), so
                // awaiting both is safe.
                let _ = tasks.dispatch.await;
                let _ = tasks.heartbeat.await;
                return;
            }
        };

        // A task ended: stop the other, release the stale writer so no
        // heartbeats are sent on the dead socket while we reconnect.
        conn_cancel.cancel();
        // Await only the task that did NOT fire above - the fired JoinHandle
        // was already consumed by `select!`.
        match ended {
            TaskEnded::Dispatch => {
                let _ = tasks.heartbeat.await;
            }
            TaskEnded::Heartbeat => {
                let _ = tasks.dispatch.await;
            }
        }
        *adapter.ws_tx.lock().await = None;

        // Brief pause before reconnecting (avoids a hot loop if the gateway
        // drops us the instant we connect).
        if sleep_or_shutdown(Duration::from_secs(1), &shutdown).await {
            return;
        }

        // Re-establish, retrying with backoff until success or shutdown.
        loop {
            conn_cancel = CancellationToken::new();
            match adapter
                .establish_connection(shutdown.clone(), conn_cancel.clone())
                .await
            {
                Ok(new_tasks) => {
                    establish_failures = 0;
                    info!("Reconnected to QQ gateway");
                    tasks = new_tasks;
                    break; // back to outer loop: wait for this connection
                }
                Err(e) => {
                    establish_failures = establish_failures.saturating_add(1);
                    let backoff = reconnect_backoff(establish_failures);
                    warn!(
                        "Reconnect attempt #{} failed: {}, retrying in {:?}",
                        establish_failures, e, backoff
                    );
                    if sleep_or_shutdown(backoff, &shutdown).await {
                        return;
                    }
                }
            }
        }
    }
}

/// Sleep for `dur`, returning early (`true`) if `shutdown` is notified.
async fn sleep_or_shutdown(dur: Duration, shutdown: &Arc<tokio::sync::Notify>) -> bool {
    tokio::select! {
        _ = tokio::time::sleep(dur) => false,
        _ = shutdown.notified() => true,
    }
}

/// Exponential backoff after `failures` consecutive establish failures:
/// 1s, 2s, 4s, 8s, 16s, 30s, 30s, ...
fn reconnect_backoff(failures: u32) -> Duration {
    let secs = 1u64 << failures.saturating_sub(1).min(5);
    Duration::from_secs(secs.min(30))
}

/// Spawn the periodic heartbeat task. Exits (handing control back to the
/// supervisor) on `conn_cancel`, `shutdown`, or a heartbeat send failure
/// (which means the connection is dead). Unlike the old version it does NOT
/// emit `Disconnected` - the supervisor detects the task ending and
/// reconnects, so the manager/agents are never torn down.
fn spawn_heartbeat(
    adapter: Arc<QqPlatformAdapter>,
    conn_cancel: CancellationToken,
    shutdown: Arc<tokio::sync::Notify>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            let interval = *adapter.heartbeat_interval.read().await;
            tokio::select! {
                _ = conn_cancel.cancelled() => {
                    debug!("Heartbeat task cancelled");
                    return;
                }
                _ = shutdown.notified() => {
                    debug!("Heartbeat task received shutdown signal");
                    return;
                }
                _ = tokio::time::sleep(interval) => {}
            }
            if let Err(e) = send_heartbeat(&adapter).await {
                warn!("Heartbeat send failed, dropping connection: {}", e);
                return;
            }
        }
    })
}

/// Send a heartbeat (op=1) carrying the last sequence number. Returns `Err`
/// if the send fails or there is no writer (connection is dead).
async fn send_heartbeat(adapter: &QqPlatformAdapter) -> std::result::Result<(), String> {
    let last_seq = *adapter.last_seq.lock().await;
    let heartbeat = GatewayPayload::heartbeat(last_seq);
    let payload = serde_json::to_string(&heartbeat).map_err(|e| e.to_string())?;
    let mut ws_tx = adapter.ws_tx.lock().await;
    match ws_tx.as_mut() {
        Some(write) => write
            .send(Message::Text(payload.into()))
            .await
            .map_err(|e| e.to_string()),
        None => Err("no WS writer (disconnected)".to_string()),
    }
}

/// Spawn the dispatch task: reads WS frames, converts dispatch events to
/// [`PlatformEvent`], and forwards them to the event channel. Exits (handing
/// control back to the supervisor) on `conn_cancel`, `shutdown`, a WS read
/// error/close, or a server-requested reconnect. Does NOT emit `Disconnected`.
fn spawn_dispatch(
    adapter: Arc<QqPlatformAdapter>,
    mut read: impl futures_util::Stream<
        Item = std::result::Result<Message, tokio_tungstenite::tungstenite::Error>,
    > + Unpin
    + Send
    + 'static,
    conn_cancel: CancellationToken,
    shutdown: Arc<tokio::sync::Notify>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = conn_cancel.cancelled() => {
                    debug!("Dispatch task cancelled");
                    return;
                }
                _ = shutdown.notified() => {
                    debug!("Dispatch task received shutdown signal");
                    return;
                }
                frame = read.next() => {
                    let msg = match frame {
                        Some(Ok(m)) => m,
                        Some(Err(e)) => {
                            warn!("WS read error, dropping connection: {}", e);
                            return;
                        }
                        None => {
                            info!("QQ dispatch stream ended");
                            return;
                        }
                    };
                    match process_frame(&adapter, msg).await {
                        FrameAction::Continue => {}
                        FrameAction::Reconnect => return,
                    }
                }
            }
        }
    })
}

/// Process a single WebSocket frame. Returns whether the connection should be
/// dropped so the supervisor can reconnect.
async fn process_frame(adapter: &QqPlatformAdapter, msg: Message) -> FrameAction {
    let text = match msg {
        Message::Text(t) => t.to_string(),
        Message::Binary(b) => String::from_utf8_lossy(&b).into_owned(),
        Message::Close(_) => {
            info!("QQ WebSocket closed by server");
            return FrameAction::Reconnect;
        }
        _ => return FrameAction::Continue,
    };

    let payload: GatewayPayload = match serde_json::from_str(&text) {
        Ok(p) => p,
        Err(e) => {
            debug!("Skipping non-JSON WS frame: {}", e);
            return FrameAction::Continue;
        }
    };

    match payload.op {
        op::HEARTBEAT_ACK => FrameAction::Continue,
        op::RECONNECT => {
            warn!("Server requested reconnect");
            FrameAction::Reconnect
        }
        op::INVALID_SESSION => {
            // Session no longer valid: clear it so the next establish does a
            // fresh Identify. (We always Identify today; this keeps session_id
            // honest for a future Resume path.)
            warn!("Invalid session, will re-identify");
            *adapter.session_id.lock().await = None;
            FrameAction::Reconnect
        }
        op::DISPATCH => {
            if let Some(seq) = payload.s {
                *adapter.last_seq.lock().await = Some(seq);
            }
            let event_name = payload.t.as_deref().unwrap_or("");
            match event_name {
                event_type::READY => {
                    info!("QQ bot is ready");
                    if let Some(d) = payload.d {
                        if let Some(sid) = d.get("session_id").and_then(|v| v.as_str()) {
                            *adapter.session_id.lock().await = Some(sid.to_string());
                        }
                    }
                }
                event_type::RESUMED => {
                    debug!("Session resumed event");
                }
                event_type::C2C_MESSAGE_CREATE | event_type::GROUP_AT_MESSAGE_CREATE => {
                    if let Some(d) = payload.d {
                        if let Ok(ev) = serde_json::from_value::<MessageEvent>(d) {
                            // Record the inbound msg_id for replies, then forward.
                            if let Some(chat_id) = chat_id_for_event(event_name, &ev) {
                                adapter.record_inbound(&chat_id, &ev.id);
                            }
                            if let Some(platform_ev) = build_platform_event(event_name, &ev) {
                                let _ = adapter.event_tx.send(platform_ev).await;
                            }
                        }
                    }
                }
                _ => {
                    debug!("Ignoring dispatch event: {}", event_name);
                }
            }
            FrameAction::Continue
        }
        _ => {
            debug!("Unhandled op {}: {:?}", payload.op, payload.t);
            FrameAction::Continue
        }
    }
}

/// Compute the platform `chat_id` for a QQ message event.
fn chat_id_for_event(event_name: &str, ev: &MessageEvent) -> Option<String> {
    match event_name {
        event_type::GROUP_AT_MESSAGE_CREATE => {
            Some(format!("group:{}", ev.group_openid.clone()?))
        }
        event_type::C2C_MESSAGE_CREATE => {
            Some(format!("private:{}", ev.user_id()?))
        }
        _ => None,
    }
}

/// Convert a QQ message event into a platform-agnostic [`PlatformEvent::Message`].
fn build_platform_event(event_name: &str, ev: &MessageEvent) -> Option<PlatformEvent> {
    let chat_id = chat_id_for_event(event_name, ev)?;
    let chat_type = match event_name {
        event_type::GROUP_AT_MESSAGE_CREATE => ChatType::Group,
        event_type::C2C_MESSAGE_CREATE => ChatType::Private,
        _ => return None,
    };
    let user_id = ev.user_id().unwrap_or("unknown").to_string();
    // QQ group @-message content typically has a leading space from the @mention.
    let mut text = ev.content.trim().to_string();

    // Convert QQ attachments to platform-agnostic media attachments.
    let attachments: Vec<MediaAttachment> = ev
        .attachments
        .iter()
        .map(|att| MediaAttachment {
            content_type: att.content_type.clone().unwrap_or_else(|| "application/octet-stream".into()),
            url: att.url.clone(),
            filename: att.filename.clone(),
            size: att.size,
            width: att.width,
            height: att.height,
        })
        .collect();

    // Append attachment descriptions to the text so the LLM knows about them.
    if !attachments.is_empty() {
        let descs: Vec<String> = attachments.iter().map(|a| a.describe()).collect();
        if text.is_empty() {
            text = descs.join("\n");
        } else {
            text = format!("{}\n{}", text, descs.join("\n"));
        }
    }

    Some(PlatformEvent::Message(ChatMessage {
        text,
        sender: SenderInfo {
            user_id,
            chat_id,
            chat_type,
        },
        attachments,
    }))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn cfg() -> QqConfig {
        QqConfig {
            app_id: "id".into(),
            app_secret: "secret".into(),
            sandbox: false,
        }
    }

    #[test]
    fn reconnect_backoff_grows_then_caps() {
        assert_eq!(reconnect_backoff(1), Duration::from_secs(1));
        assert_eq!(reconnect_backoff(2), Duration::from_secs(2));
        assert_eq!(reconnect_backoff(3), Duration::from_secs(4));
        assert_eq!(reconnect_backoff(4), Duration::from_secs(8));
        assert_eq!(reconnect_backoff(5), Duration::from_secs(16));
        // 1 << 5 = 32s, capped at 30s.
        assert_eq!(reconnect_backoff(6), Duration::from_secs(30));
        assert_eq!(reconnect_backoff(100), Duration::from_secs(30));
    }

    #[test]
    fn resolves_group_send_endpoint() {
        let (url, is_group) = resolve_send_endpoint("https://api.sgroup.qq.com", "group:abc").unwrap();
        assert_eq!(url, "https://api.sgroup.qq.com/v2/groups/abc/messages");
        assert!(is_group);
    }

    #[test]
    fn resolves_private_send_endpoint() {
        let (url, is_group) =
            resolve_send_endpoint("https://api.sgroup.qq.com", "private:user1").unwrap();
        assert_eq!(url, "https://api.sgroup.qq.com/v2/users/user1/messages");
        assert!(!is_group);
    }

    #[test]
    fn rejects_invalid_chat_id() {
        assert!(resolve_send_endpoint("https://x", "bogus").is_err());
    }

    #[test]
    fn resolves_group_upload_endpoint() {
        let (url, is_group) =
            resolve_upload_endpoint("https://api.sgroup.qq.com", "group:abc").unwrap();
        assert_eq!(
            url,
            "https://api.sgroup.qq.com/v2/groups/abc/files"
        );
        assert!(is_group);
    }

    #[test]
    fn resolves_private_upload_endpoint() {
        let (url, is_group) =
            resolve_upload_endpoint("https://api.sgroup.qq.com", "private:user1").unwrap();
        assert_eq!(
            url,
            "https://api.sgroup.qq.com/v2/users/user1/files"
        );
        assert!(!is_group);
    }

    #[test]
    fn builds_platform_event_for_group() {
        let ev = MessageEvent {
            id: "m1".into(),
            content: " hello".into(),
            author: crate::protocol::Author {
                user_openid: None,
                member_openid: Some("mem1".into()),
            },
            group_openid: Some("grp1".into()),
            attachments: vec![],
        };
        let pe = build_platform_event(event_type::GROUP_AT_MESSAGE_CREATE, &ev).unwrap();
        match pe {
            PlatformEvent::Message(m) => {
                assert_eq!(m.sender.chat_id, "group:grp1");
                assert_eq!(m.sender.chat_type, ChatType::Group);
                assert_eq!(m.text, "hello"); // leading space trimmed
                assert_eq!(m.sender.user_id, "mem1");
                assert!(m.attachments.is_empty());
            }
            _ => panic!("expected Message"),
        }
    }

    #[test]
    fn builds_platform_event_for_c2c() {
        let ev = MessageEvent {
            id: "m2".into(),
            content: "hi".into(),
            author: crate::protocol::Author {
                user_openid: Some("u1".into()),
                member_openid: None,
            },
            group_openid: None,
            attachments: vec![],
        };
        let pe = build_platform_event(event_type::C2C_MESSAGE_CREATE, &ev).unwrap();
        match pe {
            PlatformEvent::Message(m) => {
                assert_eq!(m.sender.chat_id, "private:u1");
                assert_eq!(m.sender.chat_type, ChatType::Private);
            }
            _ => panic!("expected Message"),
        }
    }

    #[test]
    fn builds_platform_event_with_attachments() {
        let ev = MessageEvent {
            id: "m3".into(),
            content: "look".into(),
            author: crate::protocol::Author {
                user_openid: Some("u2".into()),
                member_openid: None,
            },
            group_openid: None,
            attachments: vec![crate::protocol::QqAttachment {
                url: "https://cdn.qq.com/img/test.png".into(),
                content_type: Some("image/png".into()),
                filename: Some("test.png".into()),
                size: Some(204800),
                width: Some(800),
                height: Some(600),
            }],
        };
        let pe = build_platform_event(event_type::C2C_MESSAGE_CREATE, &ev).unwrap();
        match pe {
            PlatformEvent::Message(m) => {
                assert_eq!(m.attachments.len(), 1);
                assert_eq!(m.attachments[0].content_type, "image/png");
                assert_eq!(m.attachments[0].url, "https://cdn.qq.com/img/test.png");
                assert!(m.attachments[0].is_image());
                // Text should include attachment description.
                assert!(m.text.contains("用户发送了图片"));
                assert!(m.text.contains("test.png"));
            }
            _ => panic!("expected Message"),
        }
    }

    #[test]
    fn from_config_extracts_qq_section() {
        let toml_str = r#"
            [channels.qq_bot]
            app_id = "123"
            app_secret = "s"
        "#;
        // RobitConfig requires a non-empty `providers` map, so add a minimal one.
        let toml_with_providers = format!(
            "{}\n[providers.x]\nbase_url = \"https://x\"\napi_key = \"k\"\n[[providers.x.models]]\nid = \"m\"\n",
            toml_str
        );
        let config: RobitConfig = toml::from_str(&toml_with_providers).unwrap();
        let qq = QqConfig::from_config(&config).unwrap();
        assert_eq!(qq.app_id, "123");
        assert_eq!(qq.app_secret, "s");
    }

    #[test]
    fn from_config_errors_when_missing() {
        let toml_str = r#"
            [providers.x]
            base_url = "https://x"
            api_key = "k"
            [[providers.x.models]]
            id = "m"
        "#;
        let config: RobitConfig = toml::from_str(toml_str).unwrap();
        assert!(QqConfig::from_config(&config).is_err());
    }

    #[test]
    fn gateway_and_api_urls() {
        let c = cfg();
        assert!(c.gateway_url().starts_with("wss://"));
        assert!(c.api_base_url().starts_with("https://"));
    }
}