acfunlive-remake/chunks_test.go

package main

import (
	"errors"
	"math"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"
)

// 密集弹幕：每 0.02 秒一条，10 秒停留窗口里能塞进 500 条。
// 块尾扩张原来没有上限，会把窗口里的弹幕几乎全并进同一块
func denseRecords(n int) []danmuRecord {
	records := make([]danmuRecord, 0, n)
	for i := 0; i < n; i++ {
		records = append(records, danmuRecord{
			Time:    float64(i) * 0.02,
			UID:     int64(i%5 + 1),
			Name:    "密集测试",
			Content: "弹幕内容",
			Lane:    i % 10,
		})
	}
	return records
}

// 单块条数必须被硬上限挡住（续画 + 本块新出现的条数一起算），否则 -filter_complex
// 会撑爆命令行长度。
//
// 「新出现」的那些条按顺序拼起来必须正好是原始列表：续画只是同一批弹幕在后面几块里
// 接着画，不能算成新弹幕，更不能有哪一条一次都没算成新出现（那就整条丢了）
func TestSplitDanmuChunksCapsBlockSize(t *testing.T) {
	records := denseRecords(600)
	chunks := splitDanmuChunks(records, 0, 0)
	if len(chunks) < 2 {
		t.Fatalf("这么多弹幕应当分成多块，实际 %d 块", len(chunks))
	}
	fresh := make([]danmuRecord, 0, len(records))
	for i, ch := range chunks {
		if n := len(ch.records); n > maxDanmuPerChunkHard {
			t.Fatalf("第 %d 块有 %d 条弹幕，超过硬上限 %d", i+1, n, maxDanmuPerChunkHard)
		}
		if ch.carried < 0 || ch.carried > len(ch.records) {
			t.Fatalf("第 %d 块的 carried=%d 越界（这一块共 %d 条）", i+1, ch.carried, len(ch.records))
		}
		fresh = append(fresh, ch.records[ch.carried:]...)
	}
	if len(fresh) != len(records) {
		t.Fatalf("新出现的弹幕总数应当是 %d，实际 %d", len(records), len(fresh))
	}
	for i := range records {
		if fresh[i].Time != records[i].Time || fresh[i].Content != records[i].Content {
			t.Fatalf("第 %d 条新弹幕对不上：拿到 %+v，期望 %+v", i, fresh[i], records[i])
		}
	}
}

// 一条弹幕停留 10 秒，块边界经常从它中间切过去：分块器必须把它「续画」到后面几块里。
// 不续画的话每个块头屏幕上只剩刚出现的那几条，要 10 秒才重新填满 —— 观众看到的就是
// 「弹幕突然加速冲过去，然后一段时间屏幕上什么都没有」
func TestSplitDanmuChunksCarriesOnScreenDanmu(t *testing.T) {
	records := denseRecords(600) // 每 0.02 秒一条，10 秒窗口里同时有 500 条
	chunks := splitDanmuChunks(records, 0, 0)

	carriedTotal, fresh := 0, 0
	for i, ch := range chunks {
		carriedTotal += ch.carried
		fresh += len(ch.records) - ch.carried
		for _, r := range ch.records[:ch.carried] {
			// 续画只能是「上几块里已经出现过、此刻还在屏上」的那些
			if r.Time >= ch.start {
				t.Fatalf("第 %d 块（起点 %.3f）把时间 %.3f 的弹幕当成了续画，它不是上一块留下来的",
					i+1, ch.start, r.Time)
			}
			if r.Time < ch.start-danmuShowSeconds {
				t.Fatalf("第 %d 块（起点 %.3f）续画了时间 %.3f 的弹幕，它早就滚出屏了",
					i+1, ch.start, r.Time)
			}
		}
	}
	if carriedTotal == 0 {
		t.Fatal("这么密的弹幕不可能完全没有需要续画的块，续画逻辑没生效")
	}
	if fresh != len(records) {
		t.Fatalf("新出现的弹幕数不对：%d，期望 %d", fresh, len(records))
	}
}

// 块与块必须首尾相接，而且每条弹幕的 10 秒停留窗口要在所有块里被完整画满：
// 中间只要有缺口，观众就会看到那条弹幕闪一下又消失
func TestSplitDanmuChunksCoversEveryShowWindow(t *testing.T) {
	// 每 0.5 秒一条：10 秒窗口里同时有 20 条，续画远放得下，不会触发「挤掉最早的几条」
	records := make([]danmuRecord, 600)
	for i := range records {
		records[i] = danmuRecord{Time: float64(i) * 0.5, UID: 1, Name: "甲", Content: "弹幕", Lane: i % 10}
	}
	chunks := splitDanmuChunks(records, 0, 0)

	for i := 1; i < len(chunks); i++ {
		if chunks[i].start != chunks[i-1].end {
			t.Fatalf("第 %d 块和第 %d 块之间有空隙：%.3f -> %.3f",
				i, i+1, chunks[i-1].end, chunks[i].start)
		}
	}

	for _, r := range records {
		covered := r.Time
		for _, ch := range chunks {
			vs, ve, ok := danmuVisibleInChunk(r, 0, ch)
			if !ok {
				continue
			}
			if vs > covered+1e-9 {
				t.Fatalf("时间 %.3f 的弹幕在 %.3f~%.3f 之间没被画出来（前后两块没接上）",
					r.Time, covered, vs)
			}
			if ve > covered {
				covered = ve
			}
		}
		if covered < r.Time+danmuShowSeconds-1e-9 {
			t.Fatalf("时间 %.3f 的弹幕只画到 %.3f，10 秒停留窗口没画满", r.Time, covered)
		}
	}
}

// 同一时刻挤了一堆弹幕（时间戳重复）时块尾会和块头重合，那种块渲染出来一帧都没有，
// 拼进成片就是一段黑屏/缺帧
func TestSplitDanmuChunksNeverProducesEmptySpan(t *testing.T) {
	records := make([]danmuRecord, 0, 400)
	for i := 0; i < 400; i++ {
		records = append(records, danmuRecord{
			Time: 5, UID: 1, Name: "甲", Content: "同一时刻", Lane: i % 10, Width: 100,
		})
	}
	chunks := splitDanmuChunks(records, 0, 0)
	if len(chunks) < 2 {
		t.Fatalf("400 条应当分成多块，实际 %d 块", len(chunks))
	}
	for i, ch := range chunks {
		if math.IsInf(ch.end, 1) {
			continue
		}
		if ch.end <= ch.start {
			t.Fatalf("第 %d 块的时间范围是空的（%.3f ~ %.3f），渲染出来一帧都没有",
				i+1, ch.start, ch.end)
		}
	}
}

// 回退成「只合并第一个分段」时，used 必须只剩第一段：
// 自动合并收尾按 used 删源视频，返回全部就等于把没进成片的后半场删掉
func TestPrepareForMergeOnlyMarksUsedSegments(t *testing.T) {
	dir := t.TempDir()
	a := filepath.Join(dir, "a.mp4")
	b := filepath.Join(dir, "b.mp4")
	for _, p := range []string{a, b} {
		if err := os.WriteFile(p, []byte("x"), 0644); err != nil {
			t.Fatal(err)
		}
	}
	jsonl := filepath.Join(dir, "a.jsonl")
	if err := os.WriteFile(jsonl, []byte("{}\n"), 0644); err != nil {
		t.Fatal(err)
	}

	// 用一个不存在的 ffmpeg：probeDuration 拿不到时长，触发「只合并第一个分段」的回退
	segs := []recordSegment{{file: a, startNs: 0}, {file: b, startNs: int64(time.Second)}}
	video, _, _, used, cleanup, err := prepareForMerge(filepath.Join(dir, "没有这个-ffmpeg.exe"), segs, jsonl, 0)
	defer cleanup()
	if err != nil {
		t.Fatalf("回退不该报错：%v", err)
	}
	if video != a {
		t.Fatalf("回退后应当合并第一段，实际 %s", video)
	}
	if len(used) != 1 || used[0] != a {
		t.Fatalf("回退成只合并第一段时 used 应只含第一段，实际 %v", used)
	}
}

// 直接改全局任务状态：测试里只碰 stopped 这一个字段，用完恢复
func setJobStopped(t *testing.T, v bool) {
	t.Helper()
	job.Lock()
	prev := job.stopped
	job.stopped = v
	job.Unlock()
	t.Cleanup(func() {
		job.Lock()
		job.stopped = prev
		job.Unlock()
	})
}

// 现场生成一个 1 秒的小视频分段
func makeTinyVideo(t *testing.T, ffmpegFile, path, color string) {
	t.Helper()
	if err := runFFmpegWith(ffmpegFile, []string{
		"-y", "-f", "lavfi", "-i", "color=c=" + color + ":s=160x120:d=1:r=10",
		"-c:v", "libx264", "-pix_fmt", "yuv420p", path,
	}, filepath.Dir(path), nil); err != nil {
		t.Skipf("生成测试分段失败，跳过：%v", err)
	}
}

// 多段都拿到时长、成功拼接时，used 应当是全部分段（原来的删除行为不变）
func TestPrepareForMergeMarksAllSegmentsWhenJoined(t *testing.T) {
	ffmpegFile := getFFmpeg()
	if ffmpegFile == "" {
		t.Skip("找不到 ffmpeg，跳过")
	}
	setJobStopped(t, false)
	dir := t.TempDir()
	a, b := filepath.Join(dir, "a.mp4"), filepath.Join(dir, "b.mp4")
	makeTinyVideo(t, ffmpegFile, a, "red")
	makeTinyVideo(t, ffmpegFile, b, "blue")
	jsonl := filepath.Join(dir, "a.jsonl")
	if err := os.WriteFile(jsonl, []byte(`{"playResX": 160, "playResY": 120, "fontSize": 20, "startTime": 0}`+"\n"+
		`{"t": 0.2, "uid": 1, "name": "n", "avatar": "", "content": "c", "lane": 0, "width": 0}`+"\n"), 0644); err != nil {
		t.Fatal(err)
	}

	segs := []recordSegment{{file: a, startNs: 0}, {file: b, startNs: int64(time.Second)}}
	video, out, _, used, cleanup, err := prepareForMerge(ffmpegFile, segs, jsonl, 0)
	defer cleanup()
	if err != nil {
		t.Fatalf("拼接两个正常分段不该报错：%v", err)
	}
	if video != filepath.Join(dir, chunkTempPrefix+"joined.mp4") {
		t.Fatalf("应当用拼接后的文件合并，实际 %s", video)
	}
	if len(used) != 2 {
		t.Fatalf("拼接成功时 used 应包含全部分段，实际 %v", used)
	}
	if _, err := os.Stat(out); err != nil {
		t.Fatalf("重映射的弹幕文件应当存在：%v", err)
	}
	// cleanup 必须把中间文件收干净
	cleanup()
	if _, err := os.Stat(video); err == nil {
		t.Fatalf("清理后拼接出来的中间视频还在：%s", video)
	}
	if _, err := os.Stat(out); err == nil {
		t.Fatalf("清理后重映射的弹幕文件还在：%s", out)
	}
}

// 拼接中途被停止：要报成 errJobStopped，并且把拼到一半的中间文件删掉
// （留在录播目录里既占几 GB 又带 chunkTempPrefix，界面文件列表里根本看不见）
func TestPrepareForMergeDropsJoinedWhenStopped(t *testing.T) {
	ffmpegFile := getFFmpeg()
	if ffmpegFile == "" {
		t.Skip("找不到 ffmpeg，跳过")
	}
	setJobStopped(t, false)
	dir := t.TempDir()
	a, b := filepath.Join(dir, "a.mp4"), filepath.Join(dir, "b.mp4")
	makeTinyVideo(t, ffmpegFile, a, "red")
	makeTinyVideo(t, ffmpegFile, b, "blue")
	jsonl := filepath.Join(dir, "a.jsonl")
	if err := os.WriteFile(jsonl, []byte("{}\n"), 0644); err != nil {
		t.Fatal(err)
	}

	joined := filepath.Join(dir, chunkTempPrefix+"joined.mp4")
	if err := os.WriteFile(joined, []byte("拼到一半的内容"), 0644); err != nil {
		t.Fatal(err)
	}
	setJobStopped(t, true)

	segs := []recordSegment{{file: a, startNs: 0}, {file: b, startNs: int64(time.Second)}}
	_, _, _, used, cleanup, err := prepareForMerge(ffmpegFile, segs, jsonl, 0)
	defer cleanup()
	if !errors.Is(err, errJobStopped) {
		t.Fatalf("被停止时应当返回 errJobStopped，实际 %v", err)
	}
	if len(used) != 0 {
		t.Fatalf("被停止时不该把任何分段算作已合并，实际 %v", used)
	}
	if _, err := os.Stat(joined); err == nil {
		t.Fatalf("停止后拼到一半的中间文件应当被删掉：%s", joined)
	}
}

// 任务处于「已请求停止」时，拼接失败要报成 errJobStopped，而不是「拼接失败」
func TestConcatChunksReportsStopped(t *testing.T) {
	dir := t.TempDir()
	f := filepath.Join(dir, "in.mp4")
	if err := os.WriteFile(f, []byte("x"), 0644); err != nil {
		t.Fatal(err)
	}
	out := filepath.Join(dir, "out.mp4")

	setJobStopped(t, true)

	err := concatChunks(filepath.Join(dir, "没有这个-ffmpeg.exe"), []string{f}, out, dir, nil)
	if !errors.Is(err, errJobStopped) {
		t.Fatalf("停止状态下拼接失败应当返回 errJobStopped，实际 %v", err)
	}
}

// 没有停止请求时，拼接失败就是普通错误，别报成「已停止」
func TestConcatChunksReportsFailureWhenNotStopped(t *testing.T) {
	dir := t.TempDir()
	f := filepath.Join(dir, "in.mp4")
	if err := os.WriteFile(f, []byte("x"), 0644); err != nil {
		t.Fatal(err)
	}
	out := filepath.Join(dir, "out.mp4")

	setJobStopped(t, false)

	err := concatChunks(filepath.Join(dir, "没有这个-ffmpeg.exe"), []string{f}, out, dir, nil)
	if err == nil {
		t.Fatal("拼接应当失败")
	}
	if errors.Is(err, errJobStopped) {
		t.Fatalf("没有停止请求时不该报成已停止：%v", err)
	}
	if !strings.Contains(err.Error(), "拼接") {
		t.Fatalf("错误信息应当说明是拼接失败，实际 %v", err)
	}
}

// 确定性失败不该白白重试三次
func TestRunThriceSkipsNoRetry(t *testing.T) {
	calls := 0
	err := runThrice(func() error {
		calls++
		return noRetry(errors.New("确定性失败"))
	})
	if calls != 1 {
		t.Fatalf("标记为不重试的错误应当只尝试一次，实际 %d 次", calls)
	}
	if err == nil || !strings.Contains(err.Error(), "确定性失败") {
		t.Fatalf("错误应当原样返回，实际 %v", err)
	}
	if strings.Contains(err.Error(), "运行三次") {
		t.Fatalf("不该被包装成「运行三次都出现错误」：%v", err)
	}
}

// 普通错误仍然照旧重试
func TestRunThriceStillRetriesNormalError(t *testing.T) {
	calls := 0
	err := runThrice(func() error {
		calls++
		if calls == 2 {
			return nil
		}
		return errors.New("暂时失败")
	})
	if err != nil {
		t.Fatalf("第二次成功时不该报错：%v", err)
	}
	if calls != 2 {
		t.Fatalf("普通错误应当重试，实际只尝试了 %d 次", calls)
	}
}

// 交给 exec 的 ffmpeg 路径必须是绝对路径：
// Windows 上 exec 会直接拒绝执行「相对当前目录」找到的可执行文件
func TestGetFFmpegReturnsAbsPath(t *testing.T) {
	p := getFFmpeg()
	if p == "" {
		t.Skip("本机找不到 ffmpeg，跳过")
	}
	if !filepath.IsAbs(p) {
		t.Fatalf("getFFmpeg 应当返回绝对路径，实际 %q", p)
	}
}