live-sfu-demo/internal/sfu/cloudflare/client.go

package cloudflare

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"time"
)

const defaultBaseURL = "https://rtc.live.cloudflare.com/v1"

// IceServer 对应 Cloudflare Realtime session 返回的 ICE 配置（可能含 TURN 凭证）。
type IceServer struct {
	URLs       []string `json:"urls"`
	Username   string   `json:"username,omitempty"`
	Credential string   `json:"credential,omitempty"`
}

type SessionInfo struct {
	SessionID   string      `json:"sessionId"`
	AppID       string      `json:"appId"`
	RequesterIP string      `json:"requesterIp,omitempty"`
	IceServers  []IceServer `json:"iceServers,omitempty"`
}

type SessionDescription struct {
	Type string `json:"type"`
	SDP  string `json:"sdp"`
}

// TrackSpec 描述一条要加入 session 的轨道。
// location=local 表示本端发布；location=remote + SessionID 表示订阅对端 SFU 轨道。
type TrackSpec struct {
	Location                 string `json:"location"`
	SessionID                string `json:"sessionId,omitempty"`
	TrackName                string `json:"trackName,omitempty"`
	Kind                     string `json:"kind,omitempty"`
	BidirectionalMediaStream bool   `json:"bidirectionalMediaStream,omitempty"`
}

type TrackRequest struct {
	SessionDescription *SessionDescription `json:"sessionDescription,omitempty"`
	Tracks             []TrackSpec         `json:"tracks,omitempty"`
	AutoDiscover       bool                `json:"autoDiscover,omitempty"`
}

type TrackResult struct {
	TrackName        string `json:"trackName,omitempty"`
	MID              string `json:"mid,omitempty"`
	Location         string `json:"location,omitempty"`
	SessionID        string `json:"sessionId,omitempty"`
	ErrorCode        string `json:"errorCode,omitempty"`
	ErrorDescription string `json:"errorDescription,omitempty"`
}

type TracksResponse struct {
	SessionDescription             *SessionDescription `json:"sessionDescription,omitempty"`
	Tracks                         []TrackResult       `json:"tracks,omitempty"`
	RequiresImmediateRenegotiation bool                `json:"requiresImmediateRenegotiation,omitempty"`
	ErrorCode                      string              `json:"errorCode,omitempty"`
	ErrorDescription               string              `json:"errorDescription,omitempty"`
}

// Client 封装 Cloudflare Realtime REST API（服务端持有 AppSecret）。
type Client struct {
	appID      string
	appSecret  string
	baseURL    string
	httpClient *http.Client
}

func NewClient(appID, appSecret, baseURL string) *Client {
	if baseURL == "" {
		baseURL = defaultBaseURL
	}
	return &Client{
		appID:      appID,
		appSecret:  appSecret,
		baseURL:    baseURL,
		httpClient: &http.Client{Timeout: 30 * time.Second},
	}
}

func (c *Client) appsPath() string { return c.baseURL + "/apps/" + c.appID }

// CreateSession 创建一个空的 Cloudflare Realtime session（correlationId 用于房间关联）。
func (c *Client) CreateSession(correlationID string) (string, error) {
	path := c.appsPath() + "/sessions/new"
	if correlationID != "" {
		path += "?correlationId=" + url.QueryEscape(correlationID)
	}
	var resp struct {
		SessionID string `json:"sessionId"`
	}
	if err := c.doJSON(http.MethodPost, path, nil, &resp); err != nil {
		return "", err
	}
	return resp.SessionID, nil
}

func (c *Client) GetSession(sessionID string) (*SessionInfo, error) {
	var resp SessionInfo
	if err := c.doJSON(http.MethodGet, c.appsPath()+"/sessions/"+sessionID, nil, &resp); err != nil {
		return nil, err
	}
	return &resp, nil
}

// AddTracks 在 session 上发布（location=local）或订阅（location=remote）轨道，返回 SDP answer。
func (c *Client) AddTracks(sessionID string, req *TrackRequest) (*TracksResponse, error) {
	var resp TracksResponse
	if err := c.doJSON(http.MethodPost, c.appsPath()+"/sessions/"+sessionID+"/tracks/new", req, &resp); err != nil {
		return nil, err
	}
	return &resp, nil
}

// DeleteSession 终止 session 并关闭其全部轨道（用于停推 / 踢人）。
func (c *Client) DeleteSession(sessionID string) error {
	return c.doJSON(http.MethodDelete, c.appsPath()+"/sessions/"+sessionID, nil, nil)
}

func (c *Client) doJSON(method, path string, body, target interface{}) error {
	var reader io.Reader
	if body != nil {
		data, err := json.Marshal(body)
		if err != nil {
			return fmt.Errorf("cf marshal: %w", err)
		}
		reader = bytes.NewReader(data)
	}
	req, err := http.NewRequest(method, path, reader)
	if err != nil {
		return fmt.Errorf("cf build request: %w", err)
	}
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	if c.appSecret != "" {
		req.Header.Set("Authorization", "Bearer "+c.appSecret)
	}
	resp, err := c.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("cf request: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode >= 400 {
		b, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("cf api error status=%d body=%s", resp.StatusCode, string(b))
	}
	if target == nil {
		return nil
	}
	if err := json.NewDecoder(resp.Body).Decode(target); err != nil {
		return fmt.Errorf("cf decode: %w", err)
	}
	return nil
}