ggjuku/internal/app/media_resume_integration_test.go

package app

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

type mediaDisconnectWriter struct {
	http.ResponseWriter
	remaining int
}

func (writer *mediaDisconnectWriter) Write(data []byte) (int, error) {
	if len(data) > writer.remaining {
		count, _ := writer.ResponseWriter.Write(data[:writer.remaining])
		writer.remaining = 0
		return count, io.ErrUnexpectedEOF
	}
	count, err := writer.ResponseWriter.Write(data)
	writer.remaining -= count
	return count, err
}

func verifyResumedTimeline(t *testing.T, ctx context.Context, ffmpeg, input string) {
	t.Helper()
	command := exec.CommandContext(ctx, ffmpeg, "-hide_banner", "-loglevel", "warning", "-xerror", "-nostdin", "-i", input,
		"-map", "0:v:0", "-map", "0:a:0", "-progress", "pipe:1", "-f", "null", "-")
	var progress, diagnostic bytes.Buffer
	command.Stdout, command.Stderr = &progress, &diagnostic
	if err := command.Run(); err != nil {
		t.Fatalf("decode resumed timeline: %v\n%s", err, diagnostic.String())
	}
	values := make(map[string]string)
	for _, line := range strings.Split(progress.String(), "\n") {
		key, value, ok := strings.Cut(line, "=")
		if ok {
			values[key] = strings.TrimSpace(value)
		}
	}
	frames, _ := strconv.Atoi(values["frame"])
	micros, _ := strconv.Atoi(values["out_time_us"])
	if values["progress"] != "end" || frames != 1800 || micros < 59_900_000 || micros > 60_200_000 {
		t.Fatalf("interrupted media lost frames, audio or duration: %v\n%s", values, diagnostic.String())
	}
}

func TestPlaybackAndDownloadResumeInterruptedMedia(t *testing.T) {
	fixture, fixtureSession := nativePlaybackFixture(t, "60", "160x90")
	video, err := os.ReadFile(fixtureSession.tasks[0].OutPath)
	if err != nil {
		t.Fatal(err)
	}
	ffmpeg := fixture.downloader.cfg.FFmpeg
	for _, mode := range []string{"MSE", "native HLS", "prefetch", "download", "failed download"} {
		t.Run(mode, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
			defer cancel()
			var interrupted atomic.Bool
			var resumes atomic.Int32
			server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
				if request.URL.Path != "/fixture.mp4" {
					t.Error("fixture requested an unrelated resource")
					http.NotFound(writer, request)
					return
				}
				writer.Header().Set("ETag", `"generated-media-v1"`)
				if request.Header.Get("If-Range") == `"generated-media-v1"` {
					resumes.Add(1)
				}
				if request.Method == http.MethodGet && (interrupted.CompareAndSwap(false, true) || mode == "failed download") {
					writer = &mediaDisconnectWriter{ResponseWriter: writer, remaining: min(64*1024, len(video)/2)}
				}
				http.ServeContent(writer, request, "fixture.mp4", time.Time{}, bytes.NewReader(video))
			}))
			t.Cleanup(server.Close)
			app, session := prefetchFixtureApp(t)
			d := app.downloader
			d.cfg.FFmpeg, d.cfg.Retries, d.cfg.SkipBytes = ffmpeg, 3, 1
			d.client = server.Client()
			app.cfg = d.cfg
			for index := range session.tasks {
				session.tasks[index] = Task{DramaID: "hongguo:7000000000000000001", DramaTitle: "合成断流样本", Index: index + 1, Total: 3,
					OutPath: filepath.Join(t.TempDir(), "episode.mp4"),
					Chapter: Chapter{ID: fmt.Sprintf("hongguo:7000000000000000001:%d", index+1), Source: sourceHongguo, VideoURL: server.URL + "/fixture.mp4", CurrentEpisode: rawEpisode(index + 1)}}
			}
			var input string
			switch mode {
			case "native HLS":
				opened := nativeOpenFixture(t, app, 1, 0, 1)
				var address string
				if err := json.Unmarshal(opened["url"], &address); err != nil {
					t.Fatal(err)
				}
				mux := http.NewServeMux()
				app.registerPlaybackRoutes(mux)
				local := httptest.NewServer(viewerFixtureHandler(app, mux))
				t.Cleanup(local.Close)
				input = local.URL + address
			case "download", "failed download":
				var completed atomic.Bool
				err := d.DownloadEpisodeWithProgress(ctx, session.tasks[0], func(progress DownloadProgress) {
					if progress.Phase == "completed" && progress.Percent == 100 {
						completed.Store(true)
					}
				})
				if mode == "failed download" {
					_, fileErr := os.Stat(session.tasks[0].OutPath)
					if err == nil || completed.Load() || !os.IsNotExist(fileErr) {
						t.Fatal("incomplete media was marked as a completed download", err, fileErr)
					}
					return
				}
				if err != nil || !completed.Load() {
					t.Fatal("interrupted download did not complete", err)
				}
				input = session.tasks[0].OutPath
			default:
				episode := 1
				if mode == "prefetch" {
					episode = 2
					result := prefetchRequest(app, `{"session":"fixture","episode":2,"run":4,"version":1}`)
					if result.Code != http.StatusAccepted {
						t.Fatal("prefetch was not accepted", result.Body.String())
					}
					select {
					case <-session.prefetch.done:
					case <-ctx.Done():
						t.Fatal("prefetch did not complete")
					}
					if session.prefetch.view().State != "ready" {
						t.Fatal("recovered media was marked as a failed prefetch", session.prefetch.view())
					}
				}
				writer := httptest.NewRecorder()
				request := httptest.NewRequest(http.MethodGet, fmt.Sprintf("http://localhost/api/ui/playback/stream?session=fixture&episode=%d", episode), nil).WithContext(ctx)
				app.handlePlaybackStream(writer, viewerFixtureRequest(app, request))
				state, _ := app.playbackStatus("fixture", false)
				if writer.Code != http.StatusOK || state.State != "ended" || state.Error != "" || state.Duration != 60 {
					t.Fatalf("recovered playback failed: HTTP=%d state=%+v", writer.Code, state)
				}
				if mode == "prefetch" && writer.Header().Get("X-Playback-Prefetched") != "1" {
					t.Fatal("recovered prefetch was discarded")
				}
				input = filepath.Join(t.TempDir(), "resumed.mp4")
				if err := os.WriteFile(input, writer.Body.Bytes(), 0600); err != nil {
					t.Fatal(err)
				}
			}
			verifyResumedTimeline(t, ctx, ffmpeg, input)
			if !interrupted.Load() || resumes.Load() == 0 {
				t.Fatal("integration did not exercise automatic byte continuation")
			}
		})
	}
}