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)
        .on_any(wrap_any_callback!(events::any_event))
        // .on("connect", wrap_callback!(events::on_connect))
        // .on("chat.message.sendMessage", wrap_callback!(events::on_message))
        .connect()
        .await
        .unwrap();

    event!(Level::INFO, "tailchat connected");

    socket.emit("chat.converse.findAndJoinRoom", json!([])).await.unwrap();
    socket
        .emit(
            "chat.message.sendMessage",
            json!({"content": "shenbot v 0.1.0", "roomId": 1, "type": 1}),
        )
        .await
        .unwrap();

    event!(Level::INFO, "tailchat joined room");
    // 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))
                }
            }
        }
    }
}