nl-im-service/internal/mediaserver/ffmpeg_transcoder.go

/**
 * package mediaserver
 *
 * FFmpeg 实时转码器
 * 将 WebM (VP8/Opus) 转码为 FLV (H.264/AAC)
 * 用于处理不支持 H.264 编码的浏览器发送的视频流
 */
package mediaserver

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"os/exec"
	"sync"
	"time"
)

// TranscoderConfig 转码器配置
type TranscoderConfig struct {
	// 模式配置
	CopyMode bool // 是否使用 copy 模式（H.264 输入时使用，只重封装不重新编码）

	// 视频配置
	VideoCodec   string // 输出视频编解码器，默认 libx264
	VideoPreset  string // x264 预设，默认 ultrafast
	VideoBitrate string // 视频码率，默认 500k

	// 音频配置
	AudioCodec      string // 输出音频编解码器，默认 aac
	AudioBitrate    string // 音频码率，默认 64k
	AudioSampleRate int    // 音频采样率，默认 44100
	AudioChannels   int    // 音频通道数，默认 2
}

// DefaultTranscoderConfig 默认转码配置
func DefaultTranscoderConfig() *TranscoderConfig {
	return &TranscoderConfig{
		VideoCodec:      "libx264",
		VideoPreset:     "ultrafast",
		VideoBitrate:    "500k",
		AudioCodec:      "aac",
		AudioBitrate:    "64k",
		AudioSampleRate: 44100,
		AudioChannels:   2,
	}
}

// FFmpegTranscoder FFmpeg 转码器
type FFmpegTranscoder struct {
	config     *TranscoderConfig
	ffmpegPath string

	cmd    *exec.Cmd
	stdin  io.WriteCloser
	stdout io.ReadCloser
	stderr io.ReadCloser

	outputChan chan []byte
	stopChan   chan struct{}

	mu      sync.Mutex
	running bool

	// 统计信息
	inputBytes  int64
	outputBytes int64
	startTime   time.Time
}

// NewFFmpegTranscoder 创建新的转码器
func NewFFmpegTranscoder(config *TranscoderConfig) (*FFmpegTranscoder, error) {
	if config == nil {
		config = DefaultTranscoderConfig()
	}

	// 确保 FFmpeg 可用
	ffmpegPath, err := EnsureFFmpeg()
	if err != nil {
		return nil, fmt.Errorf("FFmpeg 不可用: %w", err)
	}

	return &FFmpegTranscoder{
		config:     config,
		ffmpegPath: ffmpegPath,
		outputChan: make(chan []byte, 100),
		stopChan:   make(chan struct{}),
	}, nil
}

// Start 启动转码器
func (t *FFmpegTranscoder) Start() error {
	t.mu.Lock()
	defer t.mu.Unlock()

	if t.running {
		return fmt.Errorf("转码器已在运行")
	}

	// 构建 FFmpeg 命令
	var args []string

	if t.config.CopyMode {
		// H.264 copy 模式：只重封装，不重新编码（低延迟）
		// 添加低延迟参数：禁用缓冲、减少探测时间
		args = []string{
			// 低延迟输入参数
			"-fflags", "nobuffer", // 禁用输入缓冲
			"-flags", "low_delay", // 低延迟模式
			"-probesize", "32", // 减少探测大小（字节）
			"-analyzeduration", "0", // 禁用分析时长
			"-f", "webm", // 输入格式
			"-i", "pipe:0", // 从 stdin 读取
			// 输出参数
			"-c:v", "copy", // 视频直接复制，不重新编码
			"-c:a", t.config.AudioCodec,
			"-b:a", t.config.AudioBitrate,
			"-ar", fmt.Sprintf("%d", t.config.AudioSampleRate),
			"-ac", fmt.Sprintf("%d", t.config.AudioChannels),
			// 低延迟输出参数
			"-fflags", "+genpts", // 生成时间戳
			"-f", "flv", // 输出格式
			"pipe:1", // 输出到 stdout
		}
		log.Printf("🎬 [FFmpegTranscoder] 使用 copy 模式（H.264 重封装，低延迟）")
	} else {
		// VP8/VP9 转码模式：重新编码为 H.264
		// 添加低延迟参数
		args = []string{
			// 低延迟输入参数
			"-fflags", "nobuffer", // 禁用输入缓冲
			"-flags", "low_delay", // 低延迟模式
			"-probesize", "32", // 减少探测大小（字节）
			"-analyzeduration", "0", // 禁用分析时长
			"-f", "webm", // 输入格式
			"-i", "pipe:0", // 从 stdin 读取
			// 视频编码参数
			"-c:v", t.config.VideoCodec,
			"-preset", t.config.VideoPreset,
			"-tune", "zerolatency", // 零延迟调优
			"-b:v", t.config.VideoBitrate,
			"-g", "30", // GOP 大小（减少关键帧间隔）
			"-keyint_min", "15", // 最小关键帧间隔
			// 音频编码参数
			"-c:a", t.config.AudioCodec,
			"-b:a", t.config.AudioBitrate,
			"-ar", fmt.Sprintf("%d", t.config.AudioSampleRate),
			"-ac", fmt.Sprintf("%d", t.config.AudioChannels),
			// 低延迟输出参数
			"-fflags", "+genpts", // 生成时间戳
			"-f", "flv", // 输出格式
			"pipe:1", // 输出到 stdout
		}
		log.Printf("🎬 [FFmpegTranscoder] 使用转码模式（VP8/VP9 → H.264，低延迟）")
	}

	t.cmd = exec.Command(t.ffmpegPath, args...)

	var err error

	// 获取 stdin
	t.stdin, err = t.cmd.StdinPipe()
	if err != nil {
		return fmt.Errorf("获取 stdin 失败: %w", err)
	}

	// 获取 stdout
	t.stdout, err = t.cmd.StdoutPipe()
	if err != nil {
		return fmt.Errorf("获取 stdout 失败: %w", err)
	}

	// 获取 stderr
	t.stderr, err = t.cmd.StderrPipe()
	if err != nil {
		return fmt.Errorf("获取 stderr 失败: %w", err)
	}

	// 启动 FFmpeg 进程
	if err := t.cmd.Start(); err != nil {
		return fmt.Errorf("启动 FFmpeg 失败: %w", err)
	}

	t.running = true
	t.startTime = time.Now()

	log.Printf("🎬 [FFmpegTranscoder] 转码器已启动, PID=%d", t.cmd.Process.Pid)

	// 启动输出读取协程
	go t.readOutput()

	// 启动 stderr 读取协程（用于日志）
	go t.readStderr()

	// 启动进程监控协程
	go t.monitor()

	return nil
}

// Stop 停止转码器
// 注意两个并发安全约束：
// 1. 不在此处调用 cmd.Wait()——monitor 协程是唯一的 Wait 回收方，
//    os/exec 不允许对同一 Cmd 重复 Wait（第二次行为未定义）；
// 2. 不在此处 close(outputChan)——readOutput 协程（唯一发送方）可能仍在
//    向通道发送数据，向已关闭通道发送会 panic 并直接崩溃整个进程；
//    通道统一由 readOutput 退出时关闭（"由发送方关闭"的 Go 惯例）。
func (t *FFmpegTranscoder) Stop() error {
	t.mu.Lock()
	defer t.mu.Unlock()

	if !t.running {
		return nil
	}

	t.running = false
	close(t.stopChan)

	// 关闭 stdin，通知 FFmpeg 输入结束
	if t.stdin != nil {
		t.stdin.Close()
	}

	// 杀掉进程即可：stdout 管道随之 EOF，readOutput 退出并关闭 outputChan；
	// 僵尸进程由 monitor 协程的 cmd.Wait() 统一回收
	if t.cmd != nil && t.cmd.Process != nil {
		t.cmd.Process.Kill()
	}

	duration := time.Since(t.startTime)
	log.Printf("🛑 [FFmpegTranscoder] 转码器已停止, 运行时间=%v, 输入=%d bytes, 输出=%d bytes",
		duration, t.inputBytes, t.outputBytes)

	return nil
}

// Write 写入 WebM 数据
func (t *FFmpegTranscoder) Write(data []byte) (int, error) {
	t.mu.Lock()
	if !t.running {
		t.mu.Unlock()
		return 0, fmt.Errorf("转码器未运行")
	}
	stdin := t.stdin
	t.mu.Unlock()

	n, err := stdin.Write(data)
	if err != nil {
		return n, err
	}

	t.inputBytes += int64(n)
	return n, nil
}

// Output 获取输出通道
func (t *FFmpegTranscoder) Output() <-chan []byte {
	return t.outputChan
}

// readOutput 读取 FFmpeg 输出
// 本协程是 outputChan 的唯一发送方，因此由它在退出时关闭通道：
// 既保证 range Output() 的消费者能正常结束，又杜绝"向已关闭通道发送"的 panic
func (t *FFmpegTranscoder) readOutput() {
	defer close(t.outputChan)
	buf := make([]byte, 64*1024) // 64KB 缓冲区

	for {
		select {
		case <-t.stopChan:
			return
		default:
		}

		n, err := t.stdout.Read(buf)
		if err != nil {
			if err != io.EOF {
				log.Printf("⚠️ [FFmpegTranscoder] 读取输出错误: %v", err)
			}
			return
		}

		if n > 0 {
			t.outputBytes += int64(n)

			// 复制数据并发送到通道
			data := make([]byte, n)
			copy(data, buf[:n])

			select {
			case t.outputChan <- data:
			default:
				// 通道满，丢弃数据
				log.Printf("⚠️ [FFmpegTranscoder] 输出通道满，丢弃 %d bytes", n)
			}
		}
	}
}

// readStderr 读取 FFmpeg 错误输出
func (t *FFmpegTranscoder) readStderr() {
	buf := new(bytes.Buffer)
	io.Copy(buf, t.stderr)

	if buf.Len() > 0 {
		// 只在有错误时输出
		output := buf.String()
		if len(output) > 500 {
			output = output[:500] + "..."
		}
		log.Printf("📋 [FFmpegTranscoder] FFmpeg 输出: %s", output)
	}
}

// monitor 监控 FFmpeg 进程
func (t *FFmpegTranscoder) monitor() {
	err := t.cmd.Wait()

	t.mu.Lock()
	wasRunning := t.running
	t.running = false
	t.mu.Unlock()

	if wasRunning {
		if err != nil {
			log.Printf("⚠️ [FFmpegTranscoder] FFmpeg 进程异常退出: %v", err)
		} else {
			log.Printf("ℹ️ [FFmpegTranscoder] FFmpeg 进程正常退出")
		}
	}
}

// IsRunning 检查转码器是否运行中
func (t *FFmpegTranscoder) IsRunning() bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.running
}

// Stats 获取统计信息
func (t *FFmpegTranscoder) Stats() (inputBytes, outputBytes int64, duration time.Duration) {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.inputBytes, t.outputBytes, time.Since(t.startTime)
}

// TranscoderPool 转码器池，管理多个转码器实例
type TranscoderPool struct {
	transcoders map[string]*FFmpegTranscoder
	mu          sync.RWMutex
	config      *TranscoderConfig
}

// NewTranscoderPool 创建转码器池
func NewTranscoderPool(config *TranscoderConfig) *TranscoderPool {
	if config == nil {
		config = DefaultTranscoderConfig()
	}
	return &TranscoderPool{
		transcoders: make(map[string]*FFmpegTranscoder),
		config:      config,
	}
}

// Get 获取或创建转码器
func (p *TranscoderPool) Get(streamID string) (*FFmpegTranscoder, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	if t, exists := p.transcoders[streamID]; exists && t.IsRunning() {
		return t, nil
	}

	// 创建新的转码器
	t, err := NewFFmpegTranscoder(p.config)
	if err != nil {
		return nil, err
	}

	if err := t.Start(); err != nil {
		return nil, err
	}

	p.transcoders[streamID] = t
	return t, nil
}

// Release 释放转码器
func (p *TranscoderPool) Release(streamID string) {
	p.mu.Lock()
	defer p.mu.Unlock()

	if t, exists := p.transcoders[streamID]; exists {
		t.Stop()
		delete(p.transcoders, streamID)
	}
}

// ReleaseAll 释放所有转码器
func (p *TranscoderPool) ReleaseAll() {
	p.mu.Lock()
	defer p.mu.Unlock()

	for id, t := range p.transcoders {
		t.Stop()
		delete(p.transcoders, id)
	}
}

// Count 获取活跃转码器数量
func (p *TranscoderPool) Count() int {
	p.mu.RLock()
	defer p.mu.RUnlock()
	return len(p.transcoders)
}