icalingua-python-bot/ica-rs/src/tailchat.rs

pub mod client;
pub mod events;

use futures_util::FutureExt;
use md5::{Digest, Md5};
use reqwest::{Body, ClientBuilder as reqwest_ClientBuilder};
use rust_socketio::asynchronous::{Client, ClientBuilder};
use rust_socketio::packet::PacketParser;
use rust_socketio::{Event, Payload, TransportType};
use serde_json::{from_str, from_value, json, Value};
use tracing::{event, span, Level};

use crate::config::TailchatConfig;
use crate::data_struct::tailchat::status::LoginData;
use crate::error::{ClientResult, TailchatError};
use crate::{wrap_any_callback, wrap_callback, StopGetter};

mod packet_parser {
    use bytes::{buf, Bytes};
    use rust_socketio::error::Error;
    use rust_socketio::packet::{Packet, PacketId};
    use serde::{Deserialize, Serialize};
    use serde_json::Value as JsonValue;
    use msgpack_simple::{MsgPack, MapElement}

    // #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
    // #[serde(rename = "map")]
    // struct SerdeValue {
    //     #[serde(rename = "@type")]
    //     r#type: i64,
    //     data: rmpv::Value,
    //     nsp: String,
    // }

    pub fn encode(packet: &Packet) -> Bytes {
        let raw_str = packet.data.as_ref().map(|v| v.to_string()).unwrap_or("{}".to_string());
        let raw_value: rmpv::Value = serde_json::from_str(&raw_str).unwrap();
        let value = SerdeValue {
            r#type: packet.packet_type as u8 as i64,
            data: raw_value,
            nsp: packet.nsp.clone(),
        };
        let mut buffer = rmp_serde::to_vec(&value).unwrap();
        println!("encoding packet: {:?} {:?}", packet, value);

        buffer.reverse();
        // 前面加上 \x83\xa4
        buffer.push(0xa4);
        buffer.push(0x83);
        buffer.reverse();
        Bytes::from(buffer)
    }

    pub fn default_decode(payload: &Bytes) -> Result<Packet, Error> {
        println!("decoding bytes {:?}", payload);
        todo!()
    }
}

pub async fn start_tailchat(
    config: TailchatConfig,
    stop_reciver: StopGetter,
) -> ClientResult<(), TailchatError> {
    let span = span!(Level::INFO, "Tailchat Client");
    let _enter = span.enter();

    event!(Level::INFO, "tailchat-async-rs v{} initing", crate::TAILCHAT_VERSION);

    let mut hasher = Md5::new();
    hasher.update(config.app_id.as_bytes());
    hasher.update(config.app_secret.as_bytes());

    let token = format!("{:x}", hasher.finalize());

    let mut header_map = reqwest::header::HeaderMap::new();
    header_map.append("Content-Type", "application/json".parse().unwrap());

    let client = reqwest_ClientBuilder::new().default_headers(header_map.clone()).build()?;
    let status = match client
        .post(&format!("{}/api/openapi/bot/login", config.host))
        .body(json! {{"appId": config.app_id, "token": token}}.to_string())
        .send()
        .await
    {
        Ok(resp) => {
            if resp.status().is_success() {
                let raw_data = resp.text().await?;
                let json_data = serde_json::from_str::<Value>(&raw_data).unwrap();
                let login_data = serde_json::from_value::<LoginData>(json_data["data"].clone());
                match login_data {
                    Ok(data) => data,
                    Err(e) => {
                        event!(Level::ERROR, "login failed: {}|{}", e, raw_data);
                        return Err(TailchatError::LoginFailed(e.to_string()));
                    }
                }
            } else {
                return Err(TailchatError::LoginFailed(resp.text().await?));
            }
        }
        Err(e) => return Err(TailchatError::LoginFailed(e.to_string())),
    };

    header_map.append("X-Token", status.jwt.clone().parse().unwrap());

    let packet_parser =
        PacketParser::new(Box::new(packet_parser::encode), Box::new(packet_parser::default_decode));

    let socket = ClientBuilder::new(config.host)
        .auth(json!({"token": status.jwt.clone()}))
        .transport_type(TransportType::Websocket)
        .packet_parser(packet_parser)
        .on_any(wrap_any_callback!(events::any_event))
        .on("chat.message.sendMessage", wrap_callback!(events::on_message))
        .connect()
        .await?;

    // notify:chat.message.delete
    // notify:chat.message.add

    stop_reciver.await.ok();
    event!(Level::INFO, "socketio client stopping");
    match socket.disconnect().await {
        Ok(_) => {
            event!(Level::INFO, "socketio client stopped");
            Ok(())
        }
        Err(e) => {
            // 单独处理 SocketIoError(IncompleteResponseFromEngineIo(WebsocketError(AlreadyClosed)))
            match e {
                rust_socketio::Error::IncompleteResponseFromEngineIo(inner_e) => {
                    if inner_e.to_string().contains("AlreadyClosed") {
                        event!(Level::INFO, "socketio client stopped");
                        Ok(())
                    } else {
                        event!(Level::ERROR, "socketio client stopped with error: {:?}", inner_e);
                        Err(TailchatError::SocketIoError(
                            rust_socketio::Error::IncompleteResponseFromEngineIo(inner_e),
                        ))
                    }
                }
                e => {
                    event!(Level::ERROR, "socketio client stopped with error: {}", e);
                    Err(TailchatError::SocketIoError(e))
                }
            }
        }
    }
}