Cyrene_v2/internal/capability/browser/cdp.go

package browser

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

// cdpCallTimeout 是单条 CDP 命令的兜底超时（handshake / 读超时）。正常调用还会被
// 调用方的 ctx（配置的 timeoutSeconds）更早掐断，所以它刻意给得比配置默认值宽。
const cdpCallTimeout = 60 * time.Second

// cdpMessage 是 CDP 线上信封：命令与响应共用（靠 id 关联），事件没有 id。
type cdpMessage struct {
	ID     int             `json:"id,omitempty"`
	Method string          `json:"method,omitempty"`
	Params json.RawMessage `json:"params,omitempty"`
	Result json.RawMessage `json:"result,omitempty"`
	Error  *cdpError       `json:"error,omitempty"`
}

// cdpError 是 CDP 协议级错误（例如 "Cannot find context with specified id"）。
type cdpError struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

// Error 实现 error 接口。
func (e *cdpError) Error() string { return fmt.Sprintf("CDP %d: %s", e.Code, e.Message) }

// cdpClient 是最小 CDP 客户端：一条 WebSocket、一个读协程、按 id 关联的待答表。
//
// 它**只做请求-响应**，读到的**事件一律丢弃** —— 这不是省事，而是刻意的：
// `Runtime.enable` 是本薄层唯一会「自曝自动化」的域（reference/browser-stealth/06），
// 我们从不开启它，因此既不需要事件，也不需要 `Runtime.addBinding` 那套绕法。
// evaluate 走 `Page.createIsolatedWorld` 出来的执行上下文（见 session.go）。
type cdpClient struct {
	conn    *websocket.Conn
	writeMu sync.Mutex

	mu      sync.Mutex
	nextID  int
	pending map[int]chan cdpMessage
	readErr error
	closed  chan struct{}
	once    sync.Once
}

// newCDPClient 连上 wsURL 并启动读协程。Proxy 显式置空：CDP 端点在 127.0.0.1，
// 走系统代理会连不上（本机装有代理时这是最常见的「明明在跑却连不上」）。
func newCDPClient(ctx context.Context, wsURL string) (*cdpClient, error) {
	dialer := websocket.Dialer{Proxy: nil, HandshakeTimeout: cdpCallTimeout}
	conn, resp, err := dialer.DialContext(ctx, wsURL, nil)
	if err != nil {
		status := 0
		if resp != nil {
			status = resp.StatusCode
		}
		return nil, fmt.Errorf("%w: 连接 %s: %v（HTTP %d）", ErrCDP, wsURL, err, status)
	}
	c := &cdpClient{
		conn:    conn,
		pending: make(map[int]chan cdpMessage),
		closed:  make(chan struct{}),
	}
	go c.readLoop()
	return c, nil
}

// readLoop 持续读帧直到出错：带 id 的帧派发给等待者，事件帧直接丢。
func (c *cdpClient) readLoop() {
	for {
		_, data, err := c.conn.ReadMessage()
		if err != nil {
			c.fail(fmt.Errorf("%w: 读帧: %v", ErrCDP, err))
			return
		}
		var msg cdpMessage
		if uerr := json.Unmarshal(data, &msg); uerr != nil {
			c.fail(fmt.Errorf("%w: 解码帧: %v", ErrCDP, uerr))
			return
		}
		if msg.ID == 0 {
			continue // 事件（Page.loadEventFired 之类）：本薄层不消费
		}
		c.deliver(msg)
	}
}

// deliver 把一帧响应交给它的等待者；没有等待者（已被超时放弃）就丢弃。
func (c *cdpClient) deliver(msg cdpMessage) {
	c.mu.Lock()
	ch, ok := c.pending[msg.ID]
	delete(c.pending, msg.ID)
	c.mu.Unlock()
	if ok {
		ch <- msg
	}
}

// fail 记录读协程的错误并叫醒所有等待者（连接断开后任何等待都不该继续挂）。
func (c *cdpClient) fail(err error) {
	c.once.Do(func() {
		c.mu.Lock()
		c.readErr = err
		waiters := c.pending
		c.pending = make(map[int]chan cdpMessage)
		close(c.closed)
		c.mu.Unlock()
		for _, ch := range waiters {
			close(ch)
		}
	})
}

// Call 发一条 CDP 命令并等它的结果；out 非 nil 时把 result 反序列化进去。
func (c *cdpClient) Call(ctx context.Context, method string, params any, out any) error {
	raw, err := c.call(ctx, method, params)
	if err != nil {
		return err
	}
	if out != nil && len(raw) > 0 {
		if uerr := json.Unmarshal(raw, out); uerr != nil {
			return fmt.Errorf("%w: 解码 %s 的结果: %v", ErrCDP, method, uerr)
		}
	}
	return nil
}

// call 是 Call 的内核：分配 id、登记等待、写帧、等响应或 ctx 取消。
func (c *cdpClient) call(ctx context.Context, method string, params any) (json.RawMessage, error) {
	if err := c.errIfClosed(); err != nil {
		return nil, err
	}
	id, ch, err := c.enqueue()
	if err != nil {
		return nil, err
	}
	defer c.forget(id)
	if err := c.write(cdpMessage{ID: id, Method: method, Params: mustJSON(params)}); err != nil {
		return nil, err
	}
	select {
	case msg, ok := <-ch:
		if !ok {
			return nil, c.closedErr(method)
		}
		if msg.Error != nil {
			return nil, fmt.Errorf("%w: %s: %v", ErrCDP, method, msg.Error)
		}
		return msg.Result, nil
	case <-ctx.Done():
		return nil, fmt.Errorf("%w: %s 被取消: %v", ErrCDP, method, ctx.Err())
	}
}

// enqueue 分配命令 id 与等待通道（并发安全）。
func (c *cdpClient) enqueue() (int, chan cdpMessage, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.readErr != nil {
		return 0, nil, c.readErr
	}
	c.nextID++
	id := c.nextID
	ch := make(chan cdpMessage, 1)
	c.pending[id] = ch
	return id, ch, nil
}

// forget 清掉一个等待登记（超时/取消路径上不留给读协程）。
func (c *cdpClient) forget(id int) {
	c.mu.Lock()
	delete(c.pending, id)
	c.mu.Unlock()
}

// write 串行化写帧：gorilla 的写方法不允许并发调用。
func (c *cdpClient) write(msg cdpMessage) error {
	payload, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("%w: 编码 %s: %v", ErrCDP, msg.Method, err)
	}
	c.writeMu.Lock()
	defer c.writeMu.Unlock()
	if err := c.conn.WriteMessage(websocket.TextMessage, payload); err != nil {
		return fmt.Errorf("%w: 写 %s: %v", ErrCDP, msg.Method, err)
	}
	return nil
}

// errIfClosed 报告连接是否已断（读协程记下的原始错误优先）。
func (c *cdpClient) errIfClosed() error {
	select {
	case <-c.closed:
		return c.closedErr("")
	default:
		return nil
	}
}

// closedErr 统一「连接已断」的错误，带上读协程记下的原因。
func (c *cdpClient) closedErr(method string) error {
	c.mu.Lock()
	readErr := c.readErr
	c.mu.Unlock()
	if readErr != nil {
		return readErr
	}
	return fmt.Errorf("%w: 连接已关闭（%s）", ErrCDP, method)
}

// Close 关掉连接（幂等）。
func (c *cdpClient) Close() error {
	err := c.conn.Close()
	c.fail(fmt.Errorf("%w: 连接被主动关闭", ErrCDP))
	return err
}

// mustJSON 把 params 编成 RawMessage；编码失败返回 nil（参数都是本包自造的
// map/结构体，编不出来属程序错误，交给上层按缺参数处理，绝不在这里 panic）。
func mustJSON(params any) json.RawMessage {
	if params == nil {
		return nil
	}
	raw, err := json.Marshal(params)
	if err != nil {
		return nil
	}
	return raw
}

// httpGetJSON 对 CDP 的 HTTP 端点发一次 GET 并按 out 解码（/json/list、/json/version）。
func httpGetJSON(ctx context.Context, url string, out any) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return fmt.Errorf("%w: 构造请求 %s: %v", ErrCDP, url, err)
	}
	client := &http.Client{Timeout: cdpCallTimeout}
	resp, err := client.Do(req)
	if err != nil {
		return fmt.Errorf("%w: 请求 %s: %v", ErrCDP, url, err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("%w: 请求 %s: HTTP %d", ErrCDP, url, resp.StatusCode)
	}
	if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
		return fmt.Errorf("%w: 解码 %s 的响应: %v", ErrCDP, url, err)
	}
	return nil
}