feat: initial impl for highway

Cargo.toml

@@ -12,6 +12,7 @@ md-5 = "0.11.0-pre.4"
 hex = "0.4.3"
 tracing = "0.1.41"
 tokio = { version = "1.43.0", features = [
+  "fs",
   "net",
   "io-util",
   "time",

mania/Cargo.toml

@@ -42,6 +42,8 @@ rand_chacha = "0.3.1"
 num_enum = "0.7.3"
 quick-xml = { version = "0.37.2", features = ["serialize"] }
 regex = "1.11.1"
+tokio-util = "0.7.13"
+sha1 = "0.11.0-pre.4"
 
 [build-dependencies]
 prost-build = "0.13.4"

mania/src/core/business.rs

@@ -17,6 +17,7 @@ use crate::core::connect::optimum_server;
 use crate::core::context::Context;
 use crate::core::event::prelude::*;
 use crate::core::event::{CEParse, resolve_event};
+use crate::core::highway::Highway;
 use crate::core::packet::SsoPacket;
 use crate::core::socket::{self, PacketReceiver, PacketSender};
 use crate::event::{EventDispatcher, EventListener};
@@ -125,6 +126,7 @@ impl Business {
             cache: Arc::new(Cache::new(config.cache_mode)), // TODO: construct from context
             event_dispatcher,
             event_listener,
+            highway: Arc::new(Highway::default()),
         });
 
         Ok(Self {
@@ -212,7 +214,7 @@ pub struct BusinessHandle {
     pub(crate) cache: Arc<Cache>,
     pub(crate) event_dispatcher: EventDispatcher,
     pub event_listener: EventListener,
-    // TODO: highway
+    pub(crate) highway: Arc<Highway>,
 }
 
 impl BusinessHandle {

mania/src/core/connect.rs

@@ -1,7 +1,8 @@
+use crate::core::ping::ping;
 use std::io::Result;
 use std::net::SocketAddr;
-
-use crate::core::ping::ping;
+use std::time::Duration;
+use tokio::net::TcpStream;
 
 /// Find the optimum server to connect to.
 pub async fn optimum_server(request_msf: bool, ipv6: bool) -> Result<SocketAddr> {
@@ -33,3 +34,11 @@ async fn resolve_dns(ipv6: bool) -> Result<Vec<SocketAddr>> {
     let host = tokio::net::lookup_host(host).await?;
     Ok(host.collect())
 }
+
+pub async fn tcp_connect_timeout(addr: SocketAddr, timeout: Duration) -> Result<TcpStream> {
+    let conn = tokio::net::TcpStream::connect(addr);
+    tokio::time::timeout(timeout, conn)
+        .await
+        .map_err(tokio::io::Error::from)
+        .flatten()
+}

mania/src/core/error.rs

@@ -14,6 +14,9 @@ pub enum ManiaError {
 
     #[error("An mania internal business error occurred: {0}")]
     InternalBusinessError(#[from] crate::core::business::BusinessError),
+
+    #[error("An mania internal highway error occurred: {0}")]
+    HighWayError(#[from] crate::core::highway::HighwayError),
 }
 
 pub type ManiaResult<T> = Result<T, ManiaError>;

mania/src/core/event/message/image_group_upload.rs

@@ -0,0 +1,125 @@
+use crate::core::event::prelude::*;
+use crate::core::protos::message::CustomFace;
+use crate::core::protos::service::oidb::{
+    BytesPbReserveTroop, ClientMeta, CommonHead, ExtBizInfo, FileInfo, FileType, IPv4, MsgInfo,
+    MultiMediaReqHead, NtGroupInfo, Ntv2RichMediaReq, Ntv2RichMediaResp, PicExtBizInfo,
+    PttExtBizInfo, SceneInfo, UploadInfo, UploadReq, VideoExtBizInfo,
+};
+use crate::utility::random_gen::RandomGenerator;
+
+#[derive(Debug, Default)]
+pub struct ImageGroupUploadArgs {
+    pub group_uin: u32,
+    pub size: u32,
+    pub md5: Bytes,
+    pub sha1: Bytes,
+    pub name: String,
+    pub pic_type: u32,
+    pub sub_type: u32,
+    pub height: u32,
+    pub width: u32,
+    pub summary: String,
+}
+
+#[derive(Debug, Default)]
+pub struct ImageGroupUploadRes {
+    pub msg_info: MsgInfo,
+    pub custom_face: CustomFace,
+    pub u_key: Option<String>,
+    pub ipv4s: Vec<IPv4>,
+}
+
+#[oidb_command(0x11c4, 100)]
+#[derive(Debug, ServerEvent, Default)]
+pub struct ImageGroupUploadEvent {
+    pub req: ImageGroupUploadArgs,
+    pub res: ImageGroupUploadRes,
+}
+
+impl ClientEvent for ImageGroupUploadEvent {
+    fn build(&self, _: &Context) -> CEBuildResult {
+        let req = dda!(Ntv2RichMediaReq {
+            req_head: Some(MultiMediaReqHead {
+                common: Some(CommonHead {
+                    request_id: 1,
+                    command: 100,
+                }),
+                scene: Some(dda!(SceneInfo {
+                    request_type: 2,
+                    business_type: 1,
+                    scene_type: 2,
+                    group: Some(NtGroupInfo {
+                        group_uin: self.req.group_uin,
+                    }),
+                })),
+                client: Some(ClientMeta { agent_type: 2 }),
+            }),
+            upload: Some(UploadReq {
+                upload_info: vec![UploadInfo {
+                    file_info: Some(FileInfo {
+                        file_size: self.req.size,
+                        file_hash: hex::encode(self.req.md5.clone()),
+                        file_sha1: hex::encode(self.req.sha1.clone()),
+                        file_name: self.req.name.to_owned(),
+                        r#type: Some(FileType {
+                            r#type: 1,
+                            pic_format: self.req.pic_type,
+                            video_format: 0,
+                            voice_format: 0,
+                        }),
+                        width: self.req.width,
+                        height: self.req.height,
+                        time: 0,
+                        original: 1,
+                    }),
+                    sub_file_type: 0,
+                }],
+                try_fast_upload_completed: true,
+                srv_send_msg: false,
+                client_random_id: RandomGenerator::rand_u64(),
+                compat_q_msg_scene_type: 2,
+                ext_biz_info: Some(dda!(ExtBizInfo {
+                    pic: Some(dda!(PicExtBizInfo {
+                        biz_type: self.req.sub_type,
+                        text_summary: self.req.summary.to_owned(),
+                        bytes_pb_reserve_troop: Some(dda!(BytesPbReserveTroop {
+                            sub_type: self.req.sub_type,
+                            text_summary: self.req.summary.to_owned(),
+                        })),
+                    })),
+                    video: Some(dda!(VideoExtBizInfo {
+                        bytes_pb_reserve: vec![],
+                    })),
+                    ptt: Some(dda!(PttExtBizInfo {
+                        bytes_reserve: vec![],
+                        bytes_pb_reserve: vec![],
+                        bytes_general_flags: vec![],
+                    })),
+                })),
+                client_seq: 0,
+                no_need_compat_msg: false,
+            }),
+        });
+        Ok(OidbPacket::new(0x11c4, 100, req.encode_to_vec(), false, true).to_binary())
+    }
+
+    fn parse(packet: Bytes, _: &Context) -> CEParseResult {
+        let resp = OidbPacket::parse_into::<Ntv2RichMediaResp>(packet)?;
+        let upload = resp
+            .upload
+            .ok_or_else(|| EventError::OtherError("Missing UploadResp".to_string()))?;
+        let msg_info = upload
+            .msg_info
+            .ok_or_else(|| EventError::OtherError("Missing MsgInfo".to_string()))?;
+        let custom_face = CustomFace::decode(Bytes::from(upload.compat_q_msg))?;
+        let ipv4s = upload.i_pv4s;
+        Ok(ClientResult::single(Box::new(dda!(Self {
+            res: ImageGroupUploadRes {
+                msg_info,
+                custom_face,
+                ipv4s,
+                u_key: upload.u_key,
+            }
+        }))))
+    }
+}

mania/src/core/event/message/mod.rs

@@ -2,6 +2,7 @@ pub mod file_c2c_download;
 pub mod file_group_download;
 pub mod image_c2c_download;
 pub mod image_group_download;
+pub mod image_group_upload;
 pub mod multi_msg_download;
 pub mod push_msg;
 pub mod record_c2c_download;

mania/src/core/event/message/send_message.rs

@@ -21,15 +21,7 @@ pub struct MessageResult {
 
 impl ClientEvent for SendMessageEvent {
     fn build(&self, ctx: &Context) -> CEBuildResult {
-        let packet = MessagePacker::build(
-            &self.chain,
-            &ctx.key_store
-                .uid
-                .load()
-                .as_ref()
-                .map(|arc| arc.as_ref().clone())
-                .unwrap_or_default(),
-        );
+        let packet = MessagePacker::build(&self.chain, ctx);
         Ok(BinaryPacket(packet.encode_to_vec().into()))
     }
 

mania/src/core/event/system/fetch_highway_ticket.rs

@@ -0,0 +1,38 @@
+use crate::core::event::prelude::*;
+use crate::core::protos::action::{HttpConn, HttpConn0x6ff501, HttpConn0x6ff501response};
+
+#[command("HttpConn.0x6ff_501")]
+#[derive(Debug, ServerEvent, Default)]
+pub struct FetchHighwayTicketEvent {
+    pub sig_session: Bytes,
+}
+
+impl ClientEvent for FetchHighwayTicketEvent {
+    fn build(&self, _: &Context) -> CEBuildResult {
+        let packet = HttpConn0x6ff501 {
+            http_conn: Some(dda!(HttpConn {
+                field1: 0,
+                field2: 0,
+                field3: 16,
+                field4: 1,
+                field6: 3,
+                service_types: vec![1, 5, 10, 21],
+                field9: 2,
+                field10: 9,
+                field11: 8,
+                ver: "1.0.1".to_string(),
+            })),
+        };
+        Ok(BinaryPacket(packet.encode_to_vec().into()))
+    }
+
+    fn parse(packet: Bytes, _: &Context) -> CEParseResult {
+        let res = HttpConn0x6ff501response::decode(packet)?;
+        let res = res
+            .http_conn
+            .ok_or_else(|| EventError::OtherError("No http_conn in response".to_string()))?;
+        Ok(ClientResult::single(Box::new(FetchHighwayTicketEvent {
+            sig_session: Bytes::from(res.sig_session),
+        })))
+    }
+}

mania/src/core/event/system/mod.rs

@@ -2,6 +2,7 @@ pub mod alive;
 pub mod fetch_filtered_group_request;
 pub mod fetch_friend;
 pub mod fetch_group_requests;
+pub mod fetch_highway_ticket;
 pub mod fetch_members;
 pub mod fetch_rkey;
 pub mod fetch_user_info;

mania/src/core/highway.rs

@@ -0,0 +1,58 @@
+pub mod hw_client;
+mod hw_frame_codec;
+
+use crate::core::highway::hw_client::HighwayClient;
+use crate::core::protos::service::highway::{NtHighwayDomain, NtHighwayIPv4};
+use crate::core::protos::service::oidb::IPv4;
+use arc_swap::ArcSwap;
+use bytes::Bytes;
+use std::borrow::Cow;
+use std::{io, sync::Arc};
+use thiserror::Error;
+use tokio::io::{AsyncRead, AsyncSeek};
+
+pub trait AsyncReadSeek: AsyncRead + AsyncSeek {}
+impl<T: AsyncRead + AsyncSeek> AsyncReadSeek for T {}
+pub trait AsyncPureStreamTrait: AsyncReadSeek + Send + Sync + Unpin {}
+pub type AsyncPureStream = Box<dyn AsyncReadSeek + Send + Sync + Unpin>;
+pub type AsyncStream = Arc<tokio::sync::Mutex<AsyncPureStream>>;
+
+fn int32ip2str(ip: u32) -> String {
+    let a = ip & 0xff;
+    let b = (ip >> 8) & 0xff;
+    let c = (ip >> 16) & 0xff;
+    let d = (ip >> 24) & 0xff;
+    format!("{}.{}.{}.{}", a, b, c, d)
+}
+
+pub fn oidb_ipv4s_to_highway_ipv4s(ipv4s: &[IPv4]) -> Vec<NtHighwayIPv4> {
+    ipv4s
+        .iter()
+        .map(|ip| NtHighwayIPv4 {
+            domain: Some(NtHighwayDomain {
+                is_enable: true,
+                ip: int32ip2str(ip.out_ip),
+            }),
+            port: ip.out_port,
+        })
+        .collect()
+}
+
+#[derive(Error, Debug)]
+pub enum HighwayError {
+    #[error("Invalid frame!")]
+    InvalidFrame,
+    #[error("IO error: {0}")]
+    IoError(#[from] io::Error),
+    #[error("Upload error! code={0}")]
+    UploadError(u32),
+    #[error("An error occurred in highway: {0}")]
+    OtherError(Cow<'static, str>),
+}
+
+#[derive(Default)]
+pub struct Highway {
+    pub(crate) sig_session: ArcSwap<Option<Bytes>>,
+    pub(crate) prepare_guard: tokio::sync::Mutex<()>,
+    pub(crate) client: ArcSwap<HighwayClient>,
+}