silk/server-go/internal/service/media.go

package service

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	"silk-server-go/internal/config"
)

// GbPlayResult GB28181 播放结果
type GbPlayResult struct {
	DeviceID  string
	ChannelID string
	HLS       string
	FLV       string
	WsFlv     string
	Fmp4      string
	Rtsp      string
	WebRtc    string
}

// MediaService WVP/ZLMediaKit 媒体服务代理
type MediaService struct {
	cfg            *config.Config
	httpClient     *http.Client
	cachedToken    string
	tokenExpiresAt time.Time
	mu             sync.Mutex
}

// NewMediaService 根据配置创建媒体服务
func NewMediaService(cfg *config.Config) *MediaService {
	return &MediaService{
		cfg: cfg,
		httpClient: &http.Client{
			Timeout: 15 * time.Second,
		},
	}
}

// wvpAPIBase 返回去掉末尾斜杠的 WVP API 基地址
func (m *MediaService) wvpAPIBase() string {
	return strings.TrimSuffix(m.cfg.WVPAPIBase, "/")
}

// zlmAPIBase 返回去掉末尾斜杠的 ZLMediaKit API 基地址
func (m *MediaService) zlmAPIBase() string {
	return strings.TrimSuffix(m.cfg.ZLMAPIBase, "/")
}

// login 登录 WVP API 并缓存 token（密码需 MD5），缓存 55 分钟
func (m *MediaService) login(ctx context.Context) (string, error) {
	md5Password := md5.Sum([]byte(m.cfg.WVPPassword))
	md5Hex := hex.EncodeToString(md5Password[:])

	loginURL := fmt.Sprintf("%s/api/user/login?username=%s&password=%s",
		m.wvpAPIBase(),
		m.cfg.WVPUsername,
		md5Hex,
	)

	slog.Info("登录 WVP API", "username", m.cfg.WVPUsername)

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, loginURL, nil)
	if err != nil {
		return "", fmt.Errorf("创建 WVP 登录请求失败: %w", err)
	}

	resp, err := m.httpClient.Do(req)
	if err != nil {
		return "", fmt.Errorf("WVP 登录请求失败: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("WVP 登录失败: HTTP %d", resp.StatusCode)
	}

	var body struct {
		Code int    `json:"code"`
		Msg  string `json:"msg"`
		Data struct {
			AccessToken string `json:"accessToken"`
		} `json:"data"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
		return "", fmt.Errorf("解析 WVP 登录响应失败: %w", err)
	}
	if body.Code != 0 || body.Data.AccessToken == "" {
		return "", fmt.Errorf("WVP 登录错误: %s", body.Msg)
	}

	m.mu.Lock()
	m.cachedToken = body.Data.AccessToken
	m.tokenExpiresAt = time.Now().Add(55 * time.Minute)
	m.mu.Unlock()

	slog.Info("WVP 登录成功，token 已缓存")
	return body.Data.AccessToken, nil
}

// getAuthToken 获取有效的认证 token，必要时重新登录
func (m *MediaService) getAuthToken(ctx context.Context) (string, error) {
	m.mu.Lock()
	if m.cachedToken != "" && time.Now().Before(m.tokenExpiresAt) {
		token := m.cachedToken
		m.mu.Unlock()
		return token, nil
	}
	m.mu.Unlock()
	return m.login(ctx)
}

// invalidateToken 使缓存的 token 失效
func (m *MediaService) invalidateToken() {
	m.mu.Lock()
	m.cachedToken = ""
	m.mu.Unlock()
}

// StartPlay 开始播放 GB28181 设备实时流，GET /api/play/start/{deviceId}/{channelId}
func (m *MediaService) StartPlay(deviceId, channelId string) (*GbPlayResult, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	url := fmt.Sprintf("%s/api/play/start/%s/%s", m.wvpAPIBase(), deviceId, channelId)
	slog.Info("调用 WVP play/start", "url", url)

	body, err := m.doWVPRequest(ctx, url)
	if err != nil {
		return nil, err
	}
	return m.extractPlayResult(body, deviceId, channelId), nil
}

// StopPlay 停止播放 GB28181 设备实时流，GET /api/play/stop/{deviceId}/{channelId}
func (m *MediaService) StopPlay(deviceId, channelId string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	url := fmt.Sprintf("%s/api/play/stop/%s/%s", m.wvpAPIBase(), deviceId, channelId)
	token, err := m.getAuthToken(ctx)
	if err != nil {
		return err
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return err
	}
	req.Header.Set("access-token", token)

	resp, err := m.httpClient.Do(req)
	if err != nil {
		slog.Warn("停止播放失败", "deviceId", deviceId, "channelId", channelId, "error", err)
		return nil
	}
	resp.Body.Close()
	return nil
}

// StartPlayback 开始回放 GB28181 设备历史流，GET /api/playback/start/{deviceId}/{channelId}?startTime=...&endTime=...
func (m *MediaService) StartPlayback(deviceId, channelId, startTime, endTime string) (*GbPlayResult, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	url := fmt.Sprintf("%s/api/playback/start/%s/%s?startTime=%s&endTime=%s",
		m.wvpAPIBase(), deviceId, channelId,
		startTime, endTime,
	)
	slog.Info("调用 WVP playback/start", "url", url)

	body, err := m.doWVPRequest(ctx, url)
	if err != nil {
		return nil, err
	}
	return m.extractPlayResult(body, deviceId, channelId), nil
}

// doWVPRequest 执行 WVP API 请求（token 过期时自动重试一次）
func (m *MediaService) doWVPRequest(ctx context.Context, reqURL string) (map[string]interface{}, error) {
	token, err := m.getAuthToken(ctx)
	if err != nil {
		return nil, err
	}

	body, err := m.sendWVPRequest(ctx, reqURL, token)
	if err != nil {
		return nil, err
	}

	// token 过期，重试一次
	code, ok := body["code"].(float64)
	if ok && int(code) == 401 {
		m.invalidateToken()
		newToken, err := m.getAuthToken(ctx)
		if err != nil {
			return nil, err
		}
		body, err = m.sendWVPRequest(ctx, reqURL, newToken)
		if err != nil {
			return nil, err
		}
	}

	code, ok = body["code"].(float64)
	if !ok || int(code) != 0 {
		msg, _ := body["msg"].(string)
		if msg == "" {
			msg, _ = body["message"].(string)
		}
		return nil, fmt.Errorf("WVP API 错误: %s", msg)
	}

	return body, nil
}

// sendWVPRequest 发送带 access-token 的 GET 请求
func (m *MediaService) sendWVPRequest(ctx context.Context, reqURL, token string) (map[string]interface{}, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("access-token", token)

	resp, err := m.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("流媒体服务暂时不可用: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("WVP API 返回 HTTP %d", resp.StatusCode)
	}

	var body map[string]interface{}
	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
		return nil, fmt.Errorf("解析 WVP 响应失败: %w", err)
	}
	return body, nil
}

// extractPlayResult 从 WVP 响应中提取播放地址，并将 ZLM 的 :80/ 替换为 :8081/
func (m *MediaService) extractPlayResult(body map[string]interface{}, deviceId, channelId string) *GbPlayResult {
	data, _ := body["data"].(map[string]interface{})
	if data == nil {
		data = map[string]interface{}{}
	}
	return &GbPlayResult{
		DeviceID:  deviceId,
		ChannelID: channelId,
		HLS:       fixZlmPort(getString(data, "hls")),
		FLV:       fixZlmPort(getString(data, "flv")),
		WsFlv:     fixZlmPort(getString(data, "ws_flv")),
		Fmp4:      fixZlmPort(getString(data, "fmp4")),
		Rtsp:      getString(data, "rtsp"),
		WebRtc:    fixZlmPort(getString(data, "webRtc")),
	}
}

// fixZlmPort 将 ZLMediaKit 容器内地址替换为前端可通过 server.cjs 代理访问的相对路径。
// ZLM 返回的 URL 形如 http://100.83.103.1:80/rtp/xxx.live.flv?...，
// 改为相对路径 /rtp/xxx.live.flv?... 后，前端浏览器会以同源请求走 server.cjs 代理，
// 避免 CORS / CSP 限制（server.cjs 已内置 /rtp/ 到 127.0.0.1:8081 的代理）。
func fixZlmPort(u string) string {
	if u == "" {
		return u
	}
	// 提取 /rtp/ 起始的相对路径（含 query string）
	if idx := strings.Index(u, "/rtp/"); idx >= 0 {
		return u[idx:]
	}
	// 兜底：替换 Docker 内部主机名为外部可访问 IP
	u = strings.Replace(u, "polaris-media:", "100.83.103.1:", 1)
	u = strings.Replace(u, ":80/", ":8081/", 1)
	if strings.HasSuffix(u, ":80") {
		u = u[:len(u)-2] + ":8081"
	}
	return u
}

// getString 从 map 中安全取字符串
func getString(m map[string]interface{}, key string) string {
	if v, ok := m[key].(string); ok {
		return v
	}
	return ""
}

// --- ZLMediaKit 录制控制 ---

// StartRecord 开始录制（type=1 for MP4）
func (m *MediaService) StartRecord(app, stream string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	url := fmt.Sprintf("%s/index/api/startRecord?type=1&vhost=__defaultVhost__&app=%s&stream=%s&secret=%s",
		m.zlmAPIBase(), app, stream, m.cfg.ZLMSecret,
	)
	slog.Info("ZLMediaKit startRecord", "app", app, "stream", stream)

	return m.zlmGet(ctx, url, "startRecord")
}

// StopRecord 停止录制
func (m *MediaService) StopRecord(app, stream string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	url := fmt.Sprintf("%s/index/api/stopRecord?type=1&vhost=__defaultVhost__&app=%s&stream=%s&secret=%s",
		m.zlmAPIBase(), app, stream, m.cfg.ZLMSecret,
	)
	slog.Info("ZLMediaKit stopRecord", "app", app, "stream", stream)

	return m.zlmGet(ctx, url, "stopRecord")
}

// IsRecording 检查是否正在录制
func (m *MediaService) IsRecording(app, stream string) bool {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	url := fmt.Sprintf("%s/index/api/isRecording?type=1&vhost=__defaultVhost__&app=%s&stream=%s&secret=%s",
		m.zlmAPIBase(), app, stream, m.cfg.ZLMSecret,
	)

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return false
	}
	resp, err := m.httpClient.Do(req)
	if err != nil {
		slog.Warn("ZLM isRecording 请求失败", "error", err)
		return false
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return false
	}

	var body struct {
		Code int         `json:"code"`
		Data interface{} `json:"data"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
		return false
	}
	return body.Code == 0 && body.Data == true
}

// zlmGet 执行 ZLMediaKit GET 请求并检查 code==0
func (m *MediaService) zlmGet(ctx context.Context, url, action string) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return fmt.Errorf("创建 %s 请求失败: %w", action, err)
	}
	resp, err := m.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("流媒体录制服务暂时不可用: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("ZLM %s 失败: HTTP %d", action, resp.StatusCode)
	}

	var body struct {
		Code int    `json:"code"`
		Msg  string `json:"msg"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
		return fmt.Errorf("解析 ZLM %s 响应失败: %w", action, err)
	}
	if body.Code != 0 {
		return fmt.Errorf("ZLM %s 错误: %s", action, body.Msg)
	}
	return nil
}

// WvpDeviceInfo WVP 设备同步信息（silk 与 WVP 共有参数）
type WvpDeviceInfo struct {
	Name         string
	Manufacturer string
	Transport    string
	StreamMode   string
	Password     string
	OnLine       bool
}

// SyncWvpDevices 查询 WVP 设备列表，返回 deviceId -> 设备信息映射（用于 WVP → silk 同步）
func (m *MediaService) SyncWvpDevices() (map[string]*WvpDeviceInfo, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	u := fmt.Sprintf("%s/api/device/query/devices?page=1&count=100", m.wvpAPIBase())
	body, err := m.doWVPRequest(ctx, u)
	if err != nil {
		return nil, err
	}

	result := make(map[string]*WvpDeviceInfo)
	if data, ok := body["data"].(map[string]interface{}); ok {
		if devices, ok := data["list"].([]interface{}); ok {
			for _, d := range devices {
				if dev, ok := d.(map[string]interface{}); ok {
					deviceId, _ := dev["deviceId"].(string)
					if deviceId == "" {
						continue
					}
					info := &WvpDeviceInfo{
						Name:         getString(dev, "name"),
						Manufacturer: getString(dev, "manufacturer"),
						Transport:    getString(dev, "transport"),
						StreamMode:   getString(dev, "streamMode"),
						Password:     getString(dev, "password"),
					}
					if onLine, ok := dev["onLine"].(bool); ok {
						info.OnLine = onLine
					}
					result[deviceId] = info
				}
			}
		}
	}
	return result, nil
}

// GetServerConfig 查询 WVP 服务器配置（含 SIP 参数），供前端展示
func (m *MediaService) GetServerConfig() (map[string]interface{}, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	url := fmt.Sprintf("%s/api/server/config", m.wvpAPIBase())
	body, err := m.doWVPRequest(ctx, url)
	if err != nil {
		return nil, err
	}

	if data, ok := body["data"].(map[string]interface{}); ok {
		return data, nil
	}
	return nil, fmt.Errorf("WVP 配置响应格式异常")
}

// WvpDevice WVP 设备信息
type WvpDevice struct {
	ID           int    `json:"id"`
	DeviceID     string `json:"deviceId"`
	Name         string `json:"name"`
	Manufacturer string `json:"manufacturer"`
	Model        string `json:"model"`
	OnLine       bool   `json:"onLine"`
	Transport    string `json:"transport"`
	StreamMode   string `json:"streamMode"`
	HostAddress  string `json:"hostAddress"`
}

// WvpChannel WVP 通道信息
type WvpChannel struct {
	ID        int    `json:"id"`
	DeviceID  string `json:"deviceId"`
	ChannelID string `json:"channelId"`
	Name      string `json:"name"`
	OnLine    bool   `json:"onLine"`
}

// ListWvpDevices 查询 WVP 已注册设备列表
func (m *MediaService) ListWvpDevices(query string) ([]WvpDevice, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	u := fmt.Sprintf("%s/api/device/query/devices?page=1&count=100", m.wvpAPIBase())
	if query != "" {
		u += "&query=" + url.QueryEscape(query)
	}

	body, err := m.doWVPRequest(ctx, u)
	if err != nil {
		return nil, err
	}

	var devices []WvpDevice
	if data, ok := body["data"].(map[string]interface{}); ok {
		if list, ok := data["list"].([]interface{}); ok {
			for _, d := range list {
				if dev, ok := d.(map[string]interface{}); ok {
					device := WvpDevice{
						DeviceID:     getString(dev, "deviceId"),
						Name:         getString(dev, "name"),
						Manufacturer: getString(dev, "manufacturer"),
						Model:        getString(dev, "model"),
						Transport:    getString(dev, "transport"),
						StreamMode:   getString(dev, "streamMode"),
						HostAddress:  getString(dev, "hostAddress"),
					}
					if onLine, ok := dev["onLine"].(bool); ok {
						device.OnLine = onLine
					}
					if id, ok := dev["id"].(float64); ok {
						device.ID = int(id)
					}
					devices = append(devices, device)
				}
			}
		}
	}
	return devices, nil
}

// ListWvpChannels 查询 WVP 设备的通道列表
func (m *MediaService) ListWvpChannels(deviceId string) ([]WvpChannel, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	u := fmt.Sprintf("%s/api/device/query/devices/%s/channels?page=1&count=100", m.wvpAPIBase(), url.PathEscape(deviceId))
	body, err := m.doWVPRequest(ctx, u)
	if err != nil {
		return nil, err
	}

	var channels []WvpChannel
	if data, ok := body["data"].(map[string]interface{}); ok {
		if list, ok := data["list"].([]interface{}); ok {
			for _, c := range list {
				if ch, ok := c.(map[string]interface{}); ok {
					channel := WvpChannel{
						DeviceID:  getString(ch, "deviceId"),
						ChannelID: getString(ch, "channelId"),
						Name:      getString(ch, "name"),
					}
					if onLine, ok := ch["onLine"].(bool); ok {
						channel.OnLine = onLine
					}
					if id, ok := ch["id"].(float64); ok {
						channel.ID = int(id)
					}
					channels = append(channels, channel)
				}
			}
		}
	}
	return channels, nil
}

// SyncWvpDevice 触发 WVP 设备通道同步
func (m *MediaService) SyncWvpDevice(deviceId string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	u := fmt.Sprintf("%s/api/device/query/devices/%s/sync", m.wvpAPIBase(), url.PathEscape(deviceId))
	_, err := m.doWVPRequest(ctx, u)
	return err
}

// UpdateWvpDevice 同步设备信息到 WVP（silk → WVP，共有可编辑参数：name、manufacturer、password）
func (m *MediaService) UpdateWvpDevice(deviceId string, name, manufacturer, password string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// 先查询设备获取 WVP 内部 ID
	queryURL := fmt.Sprintf("%s/api/device/query/devices/%s", m.wvpAPIBase(), url.PathEscape(deviceId))
	body, err := m.doWVPRequest(ctx, queryURL)
	if err != nil {
		return err
	}

	devData, ok := body["data"].(map[string]interface{})
	if !ok {
		return fmt.Errorf("设备尚未在 WVP 注册，请在摄像头硬件配置设备ID后等待注册完成")
	}
	devID, ok := devData["id"].(float64)
	if !ok {
		return fmt.Errorf("WVP 设备 ID 未找到")
	}

	// POST 更新设备信息（WVP API 仅接受 name/manufacturer/model 等可编辑字段）
	updateURL := fmt.Sprintf("%s/api/device/query/device/update", m.wvpAPIBase())
	token, err := m.getAuthToken(ctx)
	if err != nil {
		return err
	}

	updates := map[string]interface{}{
		"id":       int(devID),
		"deviceId": deviceId, // WVP API 要求 deviceId 字段必填
	}
	if name != "" {
		updates["name"] = name
	}
	if manufacturer != "" {
		updates["manufacturer"] = manufacturer
	}
	if password != "" {
		updates["password"] = password
	}

	updateBody, _ := json.Marshal(updates)

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, updateURL, strings.NewReader(string(updateBody)))
	if err != nil {
		return err
	}
	req.Header.Set("access-token", token)
	req.Header.Set("Content-Type", "application/json")

	resp, err := m.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("WVP 更新请求失败: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("WVP 更新失败: HTTP %d", resp.StatusCode)
	}

	// 检查 WVP 响应体中的 code 字段
	var respBody struct {
		Code int    `json:"code"`
		Msg  string `json:"msg"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&respBody); err != nil {
		return fmt.Errorf("解析 WVP 更新响应失败: %w", err)
	}
	if respBody.Code != 0 {
		return fmt.Errorf("WVP 更新失败: %s", respBody.Msg)
	}
	return nil
}

// AddWvpDevice 预先添加设备到 WVP（在摄像头注册前，设置独立密码等参数）
func (m *MediaService) AddWvpDevice(deviceId, name, manufacturer, password string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	token, err := m.getAuthToken(ctx)
	if err != nil {
		return err
	}

	addURL := fmt.Sprintf("%s/api/device/query/device/add", m.wvpAPIBase())
	payload := map[string]interface{}{
		"deviceId": deviceId,
	}
	if name != "" {
		payload["name"] = name
	}
	if manufacturer != "" {
		payload["manufacturer"] = manufacturer
	}
	if password != "" {
		payload["password"] = password
	}

	body, _ := json.Marshal(payload)
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, addURL, strings.NewReader(string(body)))
	if err != nil {
		return err
	}
	req.Header.Set("access-token", token)
	req.Header.Set("Content-Type", "application/json")

	resp, err := m.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("WVP 添加设备请求失败: %w", err)
	}
	defer resp.Body.Close()

	var respBody struct {
		Code int    `json:"code"`
		Msg  string `json:"msg"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&respBody); err != nil {
		return fmt.Errorf("解析 WVP 添加设备响应失败: %w", err)
	}
	if respBody.Code != 0 {
		return fmt.Errorf("WVP 添加设备失败: %s", respBody.Msg)
	}
	return nil
}

// DeleteWvpDevice 从 WVP 删除设备（silk 删除摄像头时同步删除 WVP 中预添加的设备）
func (m *MediaService) DeleteWvpDevice(deviceId string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	token, err := m.getAuthToken(ctx)
	if err != nil {
		return err
	}

	u := fmt.Sprintf("%s/api/device/query/devices/%s/delete", m.wvpAPIBase(), url.PathEscape(deviceId))
	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, u, nil)
	if err != nil {
		return err
	}
	req.Header.Set("access-token", token)

	resp, err := m.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("WVP 删除设备请求失败: %w", err)
	}
	defer resp.Body.Close()

	var respBody struct {
		Code int    `json:"code"`
		Msg  string `json:"msg"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&respBody); err != nil {
		return fmt.Errorf("解析 WVP 删除设备响应失败: %w", err)
	}
	if respBody.Code != 0 {
		return fmt.Errorf("WVP 删除设备失败: %s", respBody.Msg)
	}
	return nil
}