acfunlive-remake/recordjsonl_test.go

package main

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/orzogc/acfundanmu"
)

// 等替换过的后台任务全部收工。用例里触发了 fire-and-forget 的合并时，必须在用例结束前等它们跑完
func waitBackground(t *testing.T, bg *sync.WaitGroup, timeout time.Duration) {
	t.Helper()
	done := make(chan struct{})
	go func() {
		bg.Wait()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(timeout):
		t.Error("等后台任务收工超时，它可能还在读全局配置")
	}
}

// 用真正的录制写入器（danmuWriter）产生 jsonl，再按合并的流程重排一遍，
// 验证「本程序自己录下来的弹幕」合出来和以前一模一样（弹道、宽度都不变）。
// 这也是手动合并与自动合并的分水岭：录制侧写文件完全不经过 relayoutLanes，
// 只有合并渲染前会重排，所以磁盘上的 .ass/.nick.ass/.jsonl 一个字节都不变。
func TestLiveRecordingPathUnaffected(t *testing.T) {
	dir := t.TempDir()
	cfg := acfundanmu.SubConfig{Title: "测试直播", PlayResX: 1920, PlayResY: 1080,
		FontSize: 48, StartTime: 1700000000000000000}
	ass := filepath.Join(dir, "live.ass")
	w := newDanmuWriter("liveid", "stream", cfg, ass)

	texts := []string{"来啦", "这首歌好听", "主播加油啊", "打卡打卡", "哈哈哈哈哈哈哈哈哈",
		"这条弹幕稍微长一点看看弹道到底怎么排"}
	for i := 0; i < 500; i++ {
		relMs := int64(i) * 220 // 平均 0.22 秒一条，够密集
		c := &acfundanmu.Comment{Content: texts[i%len(texts)]}
		c.SendTime = (cfg.StartTime + relMs*1e6) / 1e6
		c.UserID = int64(2000 + i%9)
		c.Nickname = "观众" + string(rune('A'+i%9))
		c.Avatar = "http://example.invalid/a.png"
		w.write(c)
	}
	w.close()

	f, err := os.Open(filepath.Join(dir, "live.jsonl"))
	if err != nil {
		t.Fatal(err)
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	sc.Buffer(make([]byte, 1<<20), 1<<20)
	var meta danmuMeta
	first := true
	var recs []danmuRecord
	for sc.Scan() {
		if first {
			json.Unmarshal(sc.Bytes(), &meta)
			first = false
			continue
		}
		var r danmuRecord
		if json.Unmarshal(sc.Bytes(), &r) == nil {
			recs = append(recs, r)
		}
	}
	type snap struct{ lane, width int }
	before := make([]snap, len(recs))
	for i, r := range recs {
		before[i] = snap{r.Lane, r.Width}
	}

	relayoutLanes(recs, meta)

	laneDiff, widthDiff := 0, 0
	for i, r := range recs {
		if r.Lane != before[i].lane {
			laneDiff++
			if laneDiff <= 3 {
				t.Logf("第 %d 条弹道变了：%d → %d", i, before[i].lane, r.Lane)
			}
		}
		if r.Width != before[i].width {
			widthDiff++
		}
	}
	t.Logf("录制产生 %d 条弹幕（含空内容 %d 条），重排后弹道变化 %d 条、宽度变化 %d 条",
		len(recs), countEmpty(recs), laneDiff, widthDiff)
	if laneDiff != 0 || widthDiff != 0 {
		t.Errorf("正常录制的弹幕重排后不应有任何变化")
	}
}

func countEmpty(recs []danmuRecord) int {
	n := 0
	for _, r := range recs {
		if r.Content == "" {
			n++
		}
	}
	return n
}

// ==================== 写盘出错必须能被发现 ====================
// 弹幕是边录边写的，写盘失败（磁盘满、外接盘/网络盘掉线）如果没人管，
// 程序会照常宣告「录制完成」并自动合并 —— 出来的成片缺一大截弹幕，
// 而日志里什么都看不出来。下面几条例外情况锁住「失败要浮出来」。

// 写入失败必须被记下来，并且在 close() 时交出来
func TestDanmuWriteErrorSurfacedOnClose(t *testing.T) {
	dir := t.TempDir()
	cfg := acfundanmu.SubConfig{Title: "测试直播", PlayResX: 1920, PlayResY: 1080,
		FontSize: 48, StartTime: 1700000000000000000}
	ass := filepath.Join(dir, "live.ass")
	w := newDanmuWriter("liveid", "stream", cfg, ass)
	if w.failed() {
		t.Fatalf("刚建好的写入器不该有错误，实际：%v", w.err)
	}

	// 用「写到已关闭的句柄」来走同一条错误路径（和磁盘满、盘掉线一样，
	// 都是 os.File.Write 返回 error）。换掉之前先把原来的句柄关掉，别泄漏
	if err := w.jsonl.Close(); err != nil {
		t.Fatal(err)
	}
	dead, err := os.Open(filepath.Join(dir, "live.jsonl"))
	if err != nil {
		t.Fatal(err)
	}
	if err := dead.Close(); err != nil {
		t.Fatal(err)
	}
	w.jsonl = dead

	c := &acfundanmu.Comment{Content: "这条应该写失败"}
	c.SendTime = (cfg.StartTime + int64(500)*1e6) / 1e6
	c.UserID = 7
	c.Nickname = "观众"
	w.write(c)

	if !w.failed() {
		t.Fatal("写入失败之后 failed() 应为 true（改之前错误是被丢掉的）")
	}
	err = w.close()
	if err == nil {
		t.Fatal("close() 必须把写盘错误交出来，实际返回 nil")
	}
	if !strings.Contains(err.Error(), "jsonl") {
		t.Errorf("错误信息里要能看出是哪个文件出的问题，实际：%v", err)
	}
}

// 只记第一个错：磁盘满了之后每写一条都会失败，全记下来会把日志刷爆
func TestDanmuWriteKeepsOnlyFirstError(t *testing.T) {
	w := &danmuWriter{}
	w.fail("第一次", errors.New("磁盘满了"))
	first := w.err
	w.fail("第二次", errors.New("还是写不进去"))
	w.fail("第三次", errors.New("依旧写不进去"))
	if w.err != first {
		t.Fatalf("错误应保持在第一个，实际被后来的覆盖了：%v", w.err)
	}
	if !strings.Contains(w.err.Error(), "磁盘满了") {
		t.Errorf("应保留第一个错误的原因，实际：%v", w.err)
	}
}

// 正常录制不能被误判成失败（这是「加检查」最容易踩的反向坑）
func TestDanmuWriteNoFalseAlarm(t *testing.T) {
	dir := t.TempDir()
	cfg := acfundanmu.SubConfig{Title: "测试直播", PlayResX: 1920, PlayResY: 1080,
		FontSize: 48, StartTime: 1700000000000000000}
	w := newDanmuWriter("liveid", "stream", cfg, filepath.Join(dir, "ok.ass"))
	for i := 0; i < 30; i++ {
		c := &acfundanmu.Comment{Content: "正常弹幕"}
		c.SendTime = (cfg.StartTime + int64(i)*220*1e6) / 1e6
		c.UserID = int64(100 + i)
		c.Nickname = "观众"
		w.write(c)
	}
	if w.failed() {
		t.Fatalf("正常写入不该判失败：%v", w.err)
	}
	if err := w.close(); err != nil {
		t.Fatalf("正常写完后 close() 应返回 nil，实际：%v", err)
	}
}

// 弹幕文件写坏之后必须跳过自动合并：只是打一条日志的话，
// 照样会合成一个「缺了一大截弹幕」的成片，而且是静默的
func TestFinishPartSkipsMergeWhenDanmuBroken(t *testing.T) {
	// 这个用例会真的触发自动合并（safeGo 起的 goroutine）。它跑到用例结束之后还会读全局配置，
	// 而下一个用例开头就改配置 → -race 会判成数据竞争。所以这里把后台任务换成可等待的版本，
	// 清理时先等它收工再还原全局变量
	var bg sync.WaitGroup
	oldRun := runBackground
	runBackground = func(f func()) {
		bg.Add(1)
		go func() {
			defer bg.Done()
			f()
		}()
	}
	oldAuto, oldCtx := config.AutoMerge, mainCtx
	t.Cleanup(func() {
		waitBackground(t, &bg, 60*time.Second)
		config.AutoMerge, mainCtx = oldAuto, oldCtx
		runBackground = oldRun
	})
	config.AutoMerge = true
	mainCtx = context.Background()

	dir := t.TempDir()
	video := filepath.Join(dir, "v.mp4")
	ass := filepath.Join(dir, "v.ass")
	if err := os.WriteFile(video, []byte("fake"), 0644); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(ass, []byte("[Script Info]"), 0644); err != nil {
		t.Fatal(err)
	}

	// 测试里没跑 initialize()，这张表还没建起来
	lInfoMap.Lock()
	if lInfoMap.info == nil {
		lInfoMap.info = make(map[string]liveInfo)
	}
	lInfoMap.Unlock()

	// 对照组：一切都正常时应该触发合并
	good := "good-live-id"
	lInfoMap.Lock()
	lInfoMap.info[good] = liveInfo{LiveID: good, uid: 1, assFile: ass, firstRecordFile: video,
		recordStartNs: time.Now().UnixNano(), recordSegments: []recordSegment{{file: video, startNs: time.Now().UnixNano()}}}
	lInfoMap.Unlock()
	finishPart(good, true)
	finishPart(good, false)
	lInfoMap.RLock()
	started := lInfoMap.info[good].mergeStarted
	lInfoMap.RUnlock()
	if !started {
		t.Fatal("对照组：条件都满足时应当触发自动合并（否则这条测试本身没意义）")
	}

	// 实验组：同样条件，但弹幕写盘出过错
	bad := "bad-live-id"
	lInfoMap.Lock()
	lInfoMap.info[bad] = liveInfo{LiveID: bad, uid: 2, assFile: ass, firstRecordFile: video,
		danmuWriteFailed: true,
		recordStartNs:    time.Now().UnixNano(), recordSegments: []recordSegment{{file: video, startNs: time.Now().UnixNano()}}}
	lInfoMap.Unlock()
	finishPart(bad, true)
	finishPart(bad, false)
	lInfoMap.RLock()
	got := lInfoMap.info[bad].mergeStarted
	lInfoMap.RUnlock()
	if got {
		t.Error("弹幕文件写坏过，不该触发自动合并（否则会合成一个缺弹幕的成片）")
	}
}