lc_garage/internal/service/parking/camera_stream.go

package parking

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/gofrs/uuid/v5"
	"go.uber.org/zap"
	"wails-app/internal/dao"
	"wails-app/internal/global"
)

const (
	cameraStreamTokenTTL = 2 * time.Minute
	// A stream has to emit data recently before it is considered online. The
	// short grace window prevents an active MJPEG stream from briefly flipping
	// to offline between browser refreshes.
	cameraStreamOnlineTTL = 15 * time.Second
)

type cameraStreamToken struct {
	DeviceCode string
	ExpiresAt  time.Time
}

var cameraStreamTokens sync.Map
var cameraLastFrameAt sync.Map

// cameraMJPEGHub keeps one FFmpeg source per camera. Browser pages can create
// multiple MJPEG connections during refreshes, route transitions, or when the
// workbench is open in more than one window; those connections must not each
// open a separate RTSP session and transcoder.
var cameraMJPEGHub = newCameraStreamHub()

type cameraStreamConfig struct {
	DeviceCode string
	SourceURL  string
	FFmpegPath string
	FPS        int
	Quality    int
}

type cameraStreamSubscription struct {
	frames      <-chan []byte
	unsubscribe func()
}

type cameraStreamHub struct {
	mu       sync.Mutex
	sessions map[string]*cameraStreamSession
}

type cameraStreamSession struct {
	hub    *cameraStreamHub
	config cameraStreamConfig
	ctx    context.Context
	cancel context.CancelFunc

	mu          sync.Mutex
	nextID      uint64
	subscribers map[uint64]chan []byte
}

func newCameraStreamHub() *cameraStreamHub {
	return &cameraStreamHub{sessions: make(map[string]*cameraStreamSession)}
}

func newCameraStreamSession(hub *cameraStreamHub, config cameraStreamConfig) *cameraStreamSession {
	ctx, cancel := context.WithCancel(context.Background())
	return &cameraStreamSession{
		hub:         hub,
		config:      config,
		ctx:         ctx,
		cancel:      cancel,
		subscribers: make(map[uint64]chan []byte),
	}
}

func (h *cameraStreamHub) subscribe(config cameraStreamConfig) *cameraStreamSubscription {
	h.mu.Lock()
	session := h.sessions[config.DeviceCode]
	created := session == nil
	if session == nil {
		session = newCameraStreamSession(h, config)
		h.sessions[config.DeviceCode] = session
	}
	id, frames := session.addSubscriber()
	if created {
		go session.run()
	}
	h.mu.Unlock()
	return &cameraStreamSubscription{
		frames: frames,
		unsubscribe: func() {
			session.removeSubscriber(id)
		},
	}
}

func (h *cameraStreamHub) removeSession(session *cameraStreamSession) {
	h.mu.Lock()
	if h.sessions[session.config.DeviceCode] == session {
		delete(h.sessions, session.config.DeviceCode)
	}
	h.mu.Unlock()
}

func (h *cameraStreamHub) stop(deviceCode string) {
	h.mu.Lock()
	session := h.sessions[deviceCode]
	if session != nil {
		delete(h.sessions, deviceCode)
	}
	h.mu.Unlock()
	if session != nil {
		session.cancel()
	}
}

func (s *cameraStreamSession) addSubscriber() (uint64, <-chan []byte) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.nextID++
	frames := make(chan []byte, 64)
	s.subscribers[s.nextID] = frames
	return s.nextID, frames
}

func (s *cameraStreamSession) removeSubscriber(id uint64) {
	s.mu.Lock()
	delete(s.subscribers, id)
	empty := len(s.subscribers) == 0
	s.mu.Unlock()
	if empty {
		s.hub.removeSession(s)
		s.cancel()
	}
}

func (s *cameraStreamSession) publish(frame []byte) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, subscriber := range s.subscribers {
		select {
		case subscriber <- frame:
		default:
			// A slow client can resynchronize at the next MJPEG boundary. Never
			// let it block the camera source or every other subscribed browser.
		}
	}
}

func (s *cameraStreamSession) closeSubscribers() {
	s.mu.Lock()
	for id, subscriber := range s.subscribers {
		close(subscriber)
		delete(s.subscribers, id)
	}
	s.mu.Unlock()
}

func (s *cameraStreamSession) run() {
	defer s.closeSubscribers()
	defer s.hub.removeSession(s)

	args := []string{"-hide_banner", "-loglevel", "error"}
	if strings.HasPrefix(strings.ToLower(s.config.SourceURL), "rtsp://") {
		args = append(args, "-rtsp_transport", "tcp")
	}
	args = append(args,
		"-i", s.config.SourceURL,
		"-an",
		"-vf", fmt.Sprintf("fps=%d", s.config.FPS),
		"-q:v", fmt.Sprintf("%d", s.config.Quality),
		"-f", "mpjpeg",
		"pipe:1",
	)

	cmd := exec.CommandContext(s.ctx, s.config.FFmpegPath, args...)
	if global.GVA_LOG != nil {
		global.GVA_LOG.Info("启动共享摄像头 FFmpeg 转码", zap.String("device_code", s.config.DeviceCode), zap.String("ffmpeg", s.config.FFmpegPath), zap.String("source", redactCameraURL(s.config.SourceURL)))
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		if global.GVA_LOG != nil {
			global.GVA_LOG.Error("创建摄像头 FFmpeg 输出失败", zap.String("device_code", s.config.DeviceCode), zap.Error(err))
		}
		return
	}
	cmd.Stderr = io.Discard
	if err := cmd.Start(); err != nil {
		if global.GVA_LOG != nil {
			global.GVA_LOG.Error("摄像头 FFmpeg 启动失败", zap.String("device_code", s.config.DeviceCode), zap.String("ffmpeg", s.config.FFmpegPath), zap.Error(err))
		}
		return
	}

	buffer := make([]byte, 32*1024)
	for {
		n, readErr := stdout.Read(buffer)
		if n > 0 {
			frame := append([]byte(nil), buffer[:n]...)
			markCameraStreamFrame(s.config.DeviceCode)
			s.publish(frame)
		}
		if readErr != nil {
			if !errors.Is(readErr, io.EOF) && s.ctx.Err() == nil && global.GVA_LOG != nil {
				global.GVA_LOG.Warn("读取摄像头 FFmpeg 输出失败", zap.String("device_code", s.config.DeviceCode), zap.Error(readErr))
			}
			break
		}
	}
	if err := cmd.Wait(); err != nil && s.ctx.Err() == nil && global.GVA_LOG != nil {
		global.GVA_LOG.Error("摄像头 FFmpeg 转码进程退出", zap.String("device_code", s.config.DeviceCode), zap.Error(err))
	}
}

func markCameraStreamFrame(deviceCode string) {
	now := time.Now()
	previous, hadPrevious := cameraLastFrameAt.Load(deviceCode)
	cameraLastFrameAt.Store(deviceCode, now)
	if (!hadPrevious || now.Sub(previous.(time.Time)) > cameraStreamOnlineTTL) && global.GVA_LOG != nil {
		global.GVA_LOG.Info("摄像头视频流已在线", zap.String("device_code", deviceCode))
	}
}

func hasRecentCameraStreamFrame(deviceCode string) bool {
	value, ok := cameraLastFrameAt.Load(deviceCode)
	if !ok {
		return false
	}
	lastFrameAt, ok := value.(time.Time)
	return ok && time.Since(lastFrameAt) <= cameraStreamOnlineTTL
}

func isSystemMJPEG(protocol string) bool {
	protocol = strings.ToLower(strings.TrimSpace(protocol))
	return protocol == "system-mjpeg" || protocol == "rtsp" || protocol == "system"
}

func issueCameraStreamToken(deviceCode string) string {
	for {
		id, err := uuid.NewV4()
		if err != nil {
			continue
		}
		token := id.String()
		cameraStreamTokens.Store(token, cameraStreamToken{
			DeviceCode: deviceCode,
			ExpiresAt:  time.Now().Add(cameraStreamTokenTTL),
		})
		// 每次页面刷新都会签发新令牌；过期项只在验证时才删除，长期运行会
		// 无限累积，这里在签发时顺手清理。
		now := time.Now()
		cameraStreamTokens.Range(func(key, value any) bool {
			if entry, ok := value.(cameraStreamToken); ok && now.After(entry.ExpiresAt) {
				cameraStreamTokens.Delete(key)
			}
			return true
		})
		return token
	}
}

func validateCameraStreamToken(deviceCode, token string) bool {
	value, ok := cameraStreamTokens.Load(token)
	if !ok {
		return false
	}
	entry := value.(cameraStreamToken)
	if entry.DeviceCode != deviceCode || time.Now().After(entry.ExpiresAt) {
		cameraStreamTokens.Delete(token)
		return false
	}
	return true
}

// StreamCameraMJPEG 在系统端把摄像头原始 RTSP/HTTP 流转换为 multipart MJPEG。
// 同一摄像头的浏览器连接订阅同一个 FFmpeg 会话，避免重复转码耗尽主机资源。
func (s *PassageService) StreamCameraMJPEG(deviceCode, token string, writer http.ResponseWriter, ctx context.Context) error {
	if !validateCameraStreamToken(deviceCode, token) {
		return errors.New("摄像头播放令牌无效或已过期")
	}
	if global.GVA_DB == nil {
		return errors.New("数据库未初始化")
	}
	var camera dao.Camera
	if err := global.GVA_DB.Where("device_code = ? AND is_active = ?", deviceCode, true).First(&camera).Error; err != nil {
		return fmt.Errorf("摄像头不存在或未启用: %w", err)
	}
	if !isSystemMJPEG(camera.StreamProtocol) {
		return errors.New("当前摄像头未启用系统侧转换")
	}
	sourceURL := strings.TrimSpace(camera.SourceURL)
	if sourceURL == "" {
		sourceURL = strings.TrimSpace(camera.StreamURL)
	}
	if sourceURL == "" {
		return errors.New("摄像头未配置原始视频地址")
	}

	config := cameraStreamConfig{
		DeviceCode: deviceCode,
		SourceURL:  sourceURL,
		FFmpegPath: resolveFFmpegPath(),
		FPS:        configuredMJPEGFPS(),
		Quality:    configuredJPEGQuality(),
	}
	subscription := cameraMJPEGHub.subscribe(config)
	defer subscription.unsubscribe()

	writer.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary=ffmpeg")
	writer.Header().Set("Cache-Control", "no-store, no-cache, must-revalidate")
	writer.Header().Set("X-Content-Type-Options", "nosniff")
	if flusher, ok := writer.(http.Flusher); ok {
		flusher.Flush()
	}
	for {
		select {
		case <-ctx.Done():
			return nil
		case frame, ok := <-subscription.frames:
			if !ok {
				return nil
			}
			if _, err := writer.Write(frame); err != nil {
				return nil
			}
			if flusher, ok := writer.(http.Flusher); ok {
				flusher.Flush()
			}
		}
	}
}

func resolveFFmpegPath() string {
	ffmpegPath := strings.TrimSpace(global.GVA_CONFIG.Camera.FFmpegPath)
	if ffmpegPath == "" {
		return "ffmpeg"
	}
	if info, err := os.Stat(ffmpegPath); err == nil && info.IsDir() {
		return filepath.Join(ffmpegPath, "ffmpeg.exe")
	}
	return ffmpegPath
}

func configuredMJPEGFPS() int {
	fps := global.GVA_CONFIG.Camera.MJPEGFPS
	if fps < 1 || fps > 25 {
		return 8
	}
	return fps
}

func configuredJPEGQuality() int {
	quality := global.GVA_CONFIG.Camera.JPEGQuality
	if quality < 2 || quality > 31 {
		return 5
	}
	return quality
}

func redactCameraURL(raw string) string {
	u, err := url.Parse(raw)
	if err != nil || u.User == nil {
		return raw
	}
	return strings.Replace(raw, u.User.String()+"@", "***@", 1)
}