Files
huya-go/internal/huya/wss_client.go
T
2026-07-06 18:08:23 +08:00

675 lines
16 KiB
Go

package huya
import (
"context"
"errors"
"fmt"
"net/http"
"net/url"
"strconv"
"strings"
"sync"
"time"
"github.com/gorilla/websocket"
"huya-go/internal/taf"
"huya-go/internal/wup"
)
const (
DefaultWSSHost = "77bc035c-ws.va.huya.com"
DefaultWSSOrigin = "https://zt.huya.com"
WSSCookieUA = "webh5&0.0.1&websocket&&diypc_52775"
WSSBizUA = "web&1.0.0&huya"
WSSInitUA = "webh5&0.0.1&websocket&&h5_/index.html"
)
type WSSClient struct {
Host string
Origin string
UserAgent string
Dialer *websocket.Dialer
Logger func(string)
CallTimeout time.Duration
mu sync.Mutex
sessions map[string]*WSSSession
}
type WSSSession struct {
client *WSSClient
uid int64
cookie string
guid string
conn *websocket.Conn
writeMu sync.Mutex
pendingMu sync.Mutex
pending []chan rpcResult
closed chan struct{}
closeOnce sync.Once
closeErr error
nextWUPRequestID int32
launchGUID string
launchIP string
}
type rpcResult struct {
body []byte
err error
}
func NewWSSClient(logger func(string)) *WSSClient {
return &WSSClient{
Host: DefaultWSSHost,
Origin: DefaultWSSOrigin,
UserAgent: PCUserAgent,
Logger: logger,
CallTimeout: 10 * time.Second,
sessions: map[string]*WSSSession{},
}
}
func (c *WSSClient) logf(format string, args ...any) {
if c.Logger != nil {
c.Logger(fmt.Sprintf(format, args...))
}
}
func (c *WSSClient) Close() error {
c.mu.Lock()
sessions := make([]*WSSSession, 0, len(c.sessions))
for _, session := range c.sessions {
sessions = append(sessions, session)
}
c.sessions = map[string]*WSSSession{}
c.mu.Unlock()
var firstErr error
for _, session := range sessions {
if err := session.Close(); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
func (c *WSSClient) session(ctx context.Context, uid int64, cookie string) (*WSSSession, error) {
key := strconv.FormatInt(uid, 10)
c.mu.Lock()
if session := c.sessions[key]; session != nil && !session.isClosed() {
c.mu.Unlock()
return session, nil
}
c.mu.Unlock()
c.mu.Lock()
defer c.mu.Unlock()
if session := c.sessions[key]; session != nil && !session.isClosed() {
return session, nil
}
session, err := c.dialSession(ctx, uid, cookie)
if err != nil {
return nil, err
}
c.sessions[key] = session
return session, nil
}
func (c *WSSClient) dialSession(ctx context.Context, uid int64, cookie string) (*WSSSession, error) {
host := c.Host
if host == "" {
host = DefaultWSSHost
}
origin := c.Origin
if origin == "" {
origin = DefaultWSSOrigin
}
userAgent := c.UserAgent
if userAgent == "" {
userAgent = PCUserAgent
}
guid := CookieValue(cookie, "guid")
baseinfo, err := GenerateHTTPBaseinfo(uid, guid, NormalizeWSSCookie(cookie))
if err != nil {
return nil, err
}
wsURL := url.URL{Scheme: "wss", Host: host, Path: "/"}
wsURL.RawQuery = "baseinfo=" + baseinfo
dialer := c.Dialer
if dialer == nil {
dialer = &websocket.Dialer{
HandshakeTimeout: 10 * time.Second,
EnableCompression: true,
Proxy: http.ProxyFromEnvironment,
}
}
headers := http.Header{}
headers.Set("Origin", origin)
headers.Set("User-Agent", userAgent)
headers.Set("Accept-Language", "zh-CN,zh;q=0.9")
c.logf("[WSS] 连接 %s uid=%d", host, uid)
conn, resp, err := dialer.DialContext(ctx, wsURL.String(), headers)
if err != nil {
if resp != nil {
return nil, fmt.Errorf("wss dial: %w (status=%s)", err, resp.Status)
}
return nil, fmt.Errorf("wss dial: %w", err)
}
conn.EnableWriteCompression(true)
conn.SetReadLimit(4 << 20)
session := &WSSSession{
client: c,
uid: uid,
cookie: cookie,
guid: guid,
conn: conn,
closed: make(chan struct{}),
nextWUPRequestID: 8,
}
go session.readLoop()
go session.heartbeatLoop()
if err := session.initialize(ctx); err != nil {
_ = session.Close()
return nil, err
}
return session, nil
}
func (s *WSSSession) isClosed() bool {
select {
case <-s.closed:
return true
default:
return false
}
}
func (s *WSSSession) Close() error {
var err error
s.closeOnce.Do(func() {
close(s.closed)
err = s.conn.Close()
s.failPending(errors.New("wss closed"))
})
return err
}
func (s *WSSSession) readLoop() {
for {
messageType, raw, err := s.conn.ReadMessage()
if err != nil {
s.closeWithError(err)
return
}
if messageType != websocket.BinaryMessage {
continue
}
msg, err := DecodeWSSMessage(raw)
if err != nil {
s.client.logf("[WSS] 解码消息失败: %v", err)
continue
}
switch msg.Command {
case WSSCommandRPCResponse:
s.resolveNext(rpcResult{body: msg.Body})
case WSSCommandHeartbeatRecv:
// 心跳响应无需处理。
default:
s.client.logf("[WSS] 收到 cmd=0x%02x seq=0x%08x body=%dB", msg.Command, msg.Sequence, len(msg.Body))
}
}
}
func (s *WSSSession) heartbeatLoop() {
ticker := time.NewTicker(8 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
_ = s.sendHeartbeat(ctx)
cancel()
case <-s.closed:
return
}
}
}
func (s *WSSSession) closeWithError(err error) {
s.closeOnce.Do(func() {
s.closeErr = err
close(s.closed)
_ = s.conn.Close()
s.failPending(err)
})
}
func (s *WSSSession) failPending(err error) {
s.pendingMu.Lock()
pending := s.pending
s.pending = nil
s.pendingMu.Unlock()
for _, ch := range pending {
ch <- rpcResult{err: err}
}
}
func (s *WSSSession) resolveNext(result rpcResult) {
s.pendingMu.Lock()
if len(s.pending) == 0 {
s.pendingMu.Unlock()
return
}
ch := s.pending[0]
copy(s.pending, s.pending[1:])
s.pending = s.pending[:len(s.pending)-1]
s.pendingMu.Unlock()
ch <- result
}
func (s *WSSSession) removePending(target chan rpcResult) {
s.pendingMu.Lock()
defer s.pendingMu.Unlock()
for i, ch := range s.pending {
if ch == target {
copy(s.pending[i:], s.pending[i+1:])
s.pending = s.pending[:len(s.pending)-1]
return
}
}
}
func (s *WSSSession) writeMessage(ctx context.Context, msg WSSMessage) error {
s.writeMu.Lock()
defer s.writeMu.Unlock()
if deadline, ok := ctx.Deadline(); ok {
_ = s.conn.SetWriteDeadline(deadline)
defer s.conn.SetWriteDeadline(time.Time{})
}
return s.conn.WriteMessage(websocket.BinaryMessage, msg.Encode())
}
func (s *WSSSession) callRaw(ctx context.Context, msg WSSMessage) ([]byte, error) {
ch := make(chan rpcResult, 1)
s.pendingMu.Lock()
s.pending = append(s.pending, ch)
s.pendingMu.Unlock()
if err := s.writeMessage(ctx, msg); err != nil {
s.removePending(ch)
return nil, err
}
select {
case result := <-ch:
return result.body, result.err
case <-ctx.Done():
s.removePending(ch)
return nil, ctx.Err()
case <-s.closed:
if s.closeErr != nil {
return nil, s.closeErr
}
return nil, errors.New("wss closed")
}
}
func (s *WSSSession) initialize(ctx context.Context) error {
if err := s.sendHeartbeat(ctx); err != nil {
return err
}
if err := s.callWSLaunch(ctx); err != nil {
return err
}
if err := s.callGetConfig(ctx); err != nil {
s.client.logf("[WSS] getConfig 失败但继续: %v", err)
}
if s.launchGUID != "" {
if err := s.sendRegister(ctx, s.launchGUID); err != nil {
return err
}
}
return s.sendConfirm(ctx)
}
func (s *WSSSession) withTimeout(ctx context.Context) (context.Context, context.CancelFunc) {
if _, ok := ctx.Deadline(); ok || s.client.CallTimeout <= 0 {
return ctx, func() {}
}
return context.WithTimeout(ctx, s.client.CallTimeout)
}
func (s *WSSSession) sendHeartbeat(ctx context.Context) error {
w := taf.NewWriter()
if err := w.WriteHead(0, taf.TypeList); err != nil {
return err
}
if err := w.WriteInt32(0, 1); err != nil {
return err
}
if err := w.WriteString(0, "live:0"); err != nil {
return err
}
if err := w.WriteString(1, ""); err != nil {
return err
}
body := append(append([]byte{}, w.Bytes()...), wssTailBytes...)
return s.writeMessage(ctx, WSSMessage{Command: WSSCommandHeartbeatSend, Sequence: wssSeqHeartbeat, Body: body})
}
func (s *WSSSession) callWSLaunch(ctx context.Context) error {
req := &wsLaunchReq{}
wupReq := wup.NewRequest("launch", "wsLaunch", -1)
wupReq.Timeout = 0
if err := wupReq.WriteStruct("tReq", req); err != nil {
return err
}
data, err := wupReq.Encode()
if err != nil {
return err
}
ctx, cancel := s.withTimeout(ctx)
defer cancel()
body, err := s.callRaw(ctx, WSSMessage{
Command: WSSCommandRPCRequest,
Sequence: wssSeqWSLaunch,
Body: encodeWSSRPCBody(data),
})
if err != nil {
return err
}
var resp wup.Response
if err := resp.Decode(extractWUPFromWSSBody(body)); err != nil {
return err
}
var launch wsLaunchResp
ok, err := resp.ReadStruct("tRsp", &launch)
if err != nil {
return err
}
if !ok {
return errors.New("wsLaunch 响应缺少 tRsp")
}
s.launchGUID = launch.GUID
s.launchIP = launch.ClientIP
s.client.logf("[WSS] wsLaunch guid=%s ip=%s", launch.GUID, launch.ClientIP)
return nil
}
func (s *WSSSession) callGetConfig(ctx context.Context) error {
req := &getConfigReq{}
wupReq := wup.NewRequest("mobileui", "getConfig", 7)
wupReq.Timeout = 0
if err := wupReq.WriteStruct("tReq", req); err != nil {
return err
}
data, err := wupReq.Encode()
if err != nil {
return err
}
ctx, cancel := s.withTimeout(ctx)
defer cancel()
_, err = s.callRaw(ctx, WSSMessage{
Command: WSSCommandRPCRequest,
Sequence: wssSeqGetConfig,
Body: encodeWSSRPCBody(data),
})
return err
}
func (s *WSSSession) sendRegister(ctx context.Context, launchGUID string) error {
w := taf.NewWriter()
if err := w.WriteString(0, "HUYA&ZH&2052"); err != nil {
return err
}
if err := w.WriteString(1, launchGUID); err != nil {
return err
}
if err := w.WriteInt32(2, 0); err != nil {
return err
}
if err := w.WriteInt32(3, 0); err != nil {
return err
}
if err := w.WriteStructBegin(6); err != nil {
return err
}
if err := w.WriteMapStringString(0, nil); err != nil {
return err
}
if err := w.WriteMapStringString(1, nil); err != nil {
return err
}
if err := w.WriteInt32(3, 0); err != nil {
return err
}
if err := w.WriteStructEnd(); err != nil {
return err
}
if err := w.WriteMapStringString(7, nil); err != nil {
return err
}
if err := w.WriteInt32(8, 0); err != nil {
return err
}
body := append(append([]byte{}, w.Bytes()...), wssTailBytes...)
return s.writeMessage(ctx, WSSMessage{Command: WSSCommandRegister, Sequence: wssSeqRegister, Body: body})
}
func (s *WSSSession) sendConfirm(ctx context.Context) error {
w := taf.NewWriter()
if err := w.WriteMapStringString(0, nil); err != nil {
return err
}
body := append(append([]byte{}, w.Bytes()...), wssTailBytes...)
return s.writeMessage(ctx, WSSMessage{Command: WSSCommandConfirmSend, Sequence: wssSeqConfirm, Body: body})
}
func (s *WSSSession) nextRequestID() int32 {
id := s.nextWUPRequestID
s.nextWUPRequestID++
return id
}
func (s *WSSSession) callRPC(ctx context.Context, service, method string, reqStruct taf.Marshaler, respStruct taf.Unmarshaler) (*wup.Response, error) {
wupReq := wup.NewRequest(service, method, s.nextRequestID())
wupReq.Timeout = 0
if err := wupReq.WriteStruct("tReq", reqStruct); err != nil {
return nil, err
}
data, err := wupReq.Encode()
if err != nil {
return nil, err
}
ctx, cancel := s.withTimeout(ctx)
defer cancel()
s.client.logf("[WSS] RPC %s.%s body=%dB", service, method, len(data))
body, err := s.callRaw(ctx, WSSMessage{
Command: WSSCommandRPCRequest,
Sequence: wssSeqBusiness,
Body: encodeWSSRPCBody(data),
})
if err != nil {
return nil, err
}
var resp wup.Response
if err := resp.Decode(extractWUPFromWSSBody(body)); err != nil {
return nil, err
}
if respStruct != nil {
ok, err := resp.ReadStruct("tRsp", respStruct)
if err != nil {
return nil, err
}
if !ok {
return nil, fmt.Errorf("%s.%s 响应缺少 tRsp/tResp", service, method)
}
}
return &resp, nil
}
func (c *WSSClient) buildUser(uid int64, cookie string) ActivityUserID {
user := NewActivityUserID(uid, NormalizeWSSCookie(cookie), WSSBizUA)
user.GUID = ""
user.Token = ""
user.DeviceInfo = ""
return user
}
func (c *WSSClient) QueryUserScore(ctx context.Context, uid int64, cookie string, sid int32) (*GetUserScoreResp, error) {
session, err := c.session(ctx, uid, cookie)
if err != nil {
return nil, err
}
req := &GetUserScoreReq{User: c.buildUser(uid, cookie), SID: sid}
var resp GetUserScoreResp
_, err = session.callRPC(ctx, ServantWebActUI, FuncGetUserScore, req, &resp)
return &resp, err
}
func (c *WSSClient) GetActPrizeList(ctx context.Context, uid int64, cookie string, sid int32) (*GetActPrizeListResp, error) {
session, err := c.session(ctx, uid, cookie)
if err != nil {
return nil, err
}
req := &GetActPrizeListReq{User: c.buildUser(uid, cookie), SID: sid}
var resp GetActPrizeListResp
_, err = session.callRPC(ctx, ServantWebActUI, FuncPrizeList, req, &resp)
return &resp, err
}
func (c *WSSClient) GetUserPrizeRecords(ctx context.Context, uid int64, cookie string, sid int32) (*GetUserPrizeRecordsResp, error) {
session, err := c.session(ctx, uid, cookie)
if err != nil {
return nil, err
}
req := &GetUserPrizeRecordsReq{User: c.buildUser(uid, cookie), SID: sid}
var resp GetUserPrizeRecordsResp
_, err = session.callRPC(ctx, ServantWebActUI, FuncPrizeRecords, req, &resp)
return &resp, err
}
func (c *WSSClient) ScoreExchangePrize(ctx context.Context, uid int64, cookie string, sid, pid int32) (*ScoreExchangePrizeResp, error) {
session, err := c.session(ctx, uid, cookie)
if err != nil {
return nil, err
}
req := &ScoreExchangePrizeReq{
User: c.buildUser(uid, cookie),
SID: sid,
PID: pid,
ClientEnv: map[string]string{},
IsRole: 1,
}
var resp ScoreExchangePrizeResp
_, err = session.callRPC(ctx, ServantWebActUI, FuncExchangePrize, req, &resp)
return &resp, err
}
func NormalizeWSSCookie(cookie string) string {
cookie = NormalizeCookie(cookie)
pairs := CookiePairs(cookie)
out := []string{"huya_ua=" + WSSCookieUA}
for _, pair := range pairs {
if strings.EqualFold(pair.Key, "huya_ua") {
continue
}
out = append(out, pair.Key+"="+pair.Value)
}
return strings.Join(out, "; ")
}
type wsLaunchReq struct{}
func (r *wsLaunchReq) WriteTo(w *taf.Writer) error {
if err := w.WriteInt64(0, 0); err != nil {
return err
}
if err := w.WriteString(1, ""); err != nil {
return err
}
if err := w.WriteString(2, WSSInitUA); err != nil {
return err
}
if err := w.WriteString(3, "HUYA&ZH&2052"); err != nil {
return err
}
if err := w.WriteStructBegin(4); err != nil {
return err
}
for i := 0; i < 5; i++ {
if err := w.WriteString(i, ""); err != nil {
return err
}
}
return w.WriteStructEnd()
}
type wsLaunchResp struct {
GUID string
ClientIP string
}
func (r *wsLaunchResp) ReadFrom(rd *taf.Reader) error {
var err error
if r.GUID, err = rd.ReadString(0, false, r.GUID); err != nil {
return err
}
r.ClientIP, err = rd.ReadString(1, false, r.ClientIP)
return err
}
type getConfigReq struct{}
func (r *getConfigReq) WriteTo(w *taf.Writer) error {
if err := w.WriteStructBegin(0); err != nil {
return err
}
if err := w.WriteInt64(0, 0); err != nil {
return err
}
if err := w.WriteString(1, ""); err != nil {
return err
}
if err := w.WriteString(2, ""); err != nil {
return err
}
if err := w.WriteString(3, WSSInitUA); err != nil {
return err
}
if err := w.WriteString(4, ""); err != nil {
return err
}
if err := w.WriteInt32(5, 0); err != nil {
return err
}
if err := w.WriteString(6, ""); err != nil {
return err
}
if err := w.WriteString(7, ""); err != nil {
return err
}
if err := w.WriteStructEnd(); err != nil {
return err
}
if err := w.WriteMapStringString(1, nil); err != nil {
return err
}
if err := w.WriteString(2, "huya"); err != nil {
return err
}
if err := w.WriteHead(3, taf.TypeList); err != nil {
return err
}
return w.WriteInt32(0, 0)
}