优化

internal/handler/reader_media_test.go

@@ -3,6 +3,7 @@ package handler
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"strconv"
@@ -10,6 +11,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/gin-gonic/gin"
 
@@ -364,9 +366,16 @@ func TestReaderAudioCacheServesFromDisk(t *testing.T) {
 	}
 }
 
-// TestReaderAudioCacheRejectsTruncatedUpstream 音轨拉不完整时重试；全部失败回 502，
-// 且不能留下半截缓存文件（否则后续会把截断的音频当完整文件下发）。
-func TestReaderAudioCacheRejectsTruncatedUpstream(t *testing.T) {
+// TestReaderAudioCacheAbortsOnTruncatedUpstream 首播直通模式下上游持续截断的契约：
+//
+// 冷启动走边下边发（见 handleStreamingReaderAudio），响应头和前缀字节在探测
+// 通过后就已发出，因此中途截断不可能再改写成 502 —— 必须中止连接让客户端
+// 判定「传输不完整」，同时保留 .part 分片。
+//
+// 关键不变量：**绝不能留下可命中的完整缓存**（否则后续会把截断音频当真完整
+// 文件下发，播放器表现为「时长正确但没有声音」）。下一次请求带着分片走续传
+// 路径，那时才可能给出 502 + 明确文案。
+func TestReaderAudioCacheAbortsOnTruncatedUpstream(t *testing.T) {
 	const declared = 400 << 10
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "audio/mp4")
@@ -391,15 +400,43 @@ func TestReaderAudioCacheRejectsTruncatedUpstream(t *testing.T) {
 	// 无后缀地址 + 音频书籍 → 走音轨落盘缓存。
 	signed := svc.Reader.ProxyURL(book.ID, srv.URL+"/media")
 
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, signed, nil)
+	// panic(http.ErrAbortHandler) 由 net/http 捕获；httptest 下会向上冒泡，
+	// 这里断言它确实发生了（即「没有静默地把半截音频当成功」）。
+	func() {
+		defer func() {
+			recovered := recover()
+			if recovered == nil {
+				t.Fatalf("上游截断时必须中止响应，实际正常返回了 %d 字节", w.Body.Len())
+			}
+			if recovered != http.ErrAbortHandler {
+				t.Fatalf("中止响应时应使用 http.ErrAbortHandler，实际 %v", recovered)
+			}
+		}()
+		router.ServeHTTP(w, req)
+	}()
+
+	if w.Body.Len() >= declared {
+		t.Fatalf("不应交付完整长度的响应体（上游只写了 128KB）")
+	}
+	if _, _, ok := svc.Reader.OpenCachedRemoteAudio(book.ID, srv.URL+"/media"); ok {
+		t.Fatalf("中断的下载不能留下可命中的完整缓存")
+	}
+	if !svc.Reader.HasPartialRemoteAudio(book.ID, srv.URL+"/media") {
+		t.Fatalf("中断后应保留 .part 分片供续传")
+	}
+
+	// 二次请求带分片走续传路径；上游依旧忽略 Range 并截断 → 这时才给 502。
 	code, body := getReaderMedia(t, router, signed)
 	if code != http.StatusBadGateway {
-		t.Fatalf("上游持续截断应当返回 502，实际 %d body=%.120s", code, body)
+		t.Fatalf("续传路径持续失败应返回 502，实际 %d body=%.120s", code, body)
 	}
 	if strings.Contains(body, "ftyp") {
 		t.Fatalf("不能把半截音频交给客户端: %q", body)
 	}
 	if _, _, ok := svc.Reader.OpenCachedRemoteAudio(book.ID, srv.URL+"/media"); ok {
-		t.Fatalf("失败的下载不应留下缓存文件")
+		t.Fatalf("失败的续传仍不应留下完整缓存")
 	}
 }
 
@@ -422,7 +459,7 @@ func getReaderMedia(t *testing.T, router *gin.Engine, signedPath string) (int, s
 // primeTicketBook 先读一次章节（签出第一张票并落缓存），返回该音轨的签名代理路径。
 func primeTicketBook(t *testing.T, svc *service.Container, bookID string) string {
 	t.Helper()
-	out, err := svc.Reader.GetContentForBook(t.Context(), "u1", bookID, 0)
+	out, err := svc.Reader.GetContentForBook(t.Context(), "u1", bookID, 0, reader.ContentOptions{})
 	if err != nil {
 		t.Fatalf("首次取章节失败: %v", err)
 	}
@@ -631,3 +668,97 @@ func prepareMediaTicketBook(t *testing.T) (*service.Container, *mediaTicketServe
 	}
 	return svc, srv, book
 }
+
+// TestReaderAudioStreamsFirstByteBeforeDownloadCompletes 首播 TTFB 回归：
+// 冷启动必须边下边发，而不是等整章下完再一次性下发。
+//
+// 上游在第一块之后卡住不放：直通实现此时应已把那块转给客户端，所以读端能
+// 拿到字节；旧实现（整段下载后才 ServeContent）在 release 关闭前一个字节都
+// 不会发出，读端会一直阻塞。
+func TestReaderAudioStreamsFirstByteBeforeDownloadCompletes(t *testing.T) {
+	// 总长必须大于探测窗口（readerMediaProbeBytes=64KB）：首播要先读完这一小段
+	// 才能判断上游给的是错误页还是媒体，这之后才可能外发。卡点在探测结束之后，
+	// 才能证明「整段下完才发」和「边下边发」的区别。
+	const chunks = 16
+	const chunkBytes = 16 << 10
+	const stallAfter = 4 // 写满探测窗口（4×16KB=64KB）后卡住
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	unblock := func() { releaseOnce.Do(func() { close(release) }) }
+	defer unblock()
+
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "audio/mp4")
+		w.Header().Set("Content-Length", strconv.Itoa(chunks*chunkBytes))
+		w.WriteHeader(http.StatusOK)
+		for i := 0; i < chunks; i++ {
+			if _, err := w.Write(bytes.Repeat([]byte("m4a!"), chunkBytes/4)); err != nil {
+				return
+			}
+			w.(http.Flusher).Flush()
+			if i+1 == stallAfter {
+				// 探测窗口刚写满就卡住，直到测试确认客户端已经收到字节。
+				<-release
+			}
+		}
+	}))
+	defer upstream.Close()
+
+	svc, _, book := prepareMediaTicketBookWithUpstream(t, upstream.URL)
+	signed := svc.Reader.ProxyURL(book.ID, upstream.URL+"/media")
+
+	gin.SetMode(gin.TestMode)
+	router := gin.New()
+	router.GET("/api/reader/media", readerMediaProxyHandler(svc))
+	proxy := httptest.NewServer(router)
+	defer proxy.Close()
+
+	resp, err := http.Get(proxy.URL + signed)
+	if err != nil {
+		t.Fatalf("请求媒体代理失败: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("直通应返回 200，实际 %d", resp.StatusCode)
+	}
+
+	// 上游仍卡在 release 上：此时能读到第一批字节，才说明是边下边发。
+	first := make([]byte, 4)
+	readDone := make(chan error, 1)
+	go func() {
+		_, readErr := io.ReadFull(resp.Body, first)
+		readDone <- readErr
+	}()
+	select {
+	case readErr := <-readDone:
+		if readErr != nil {
+			t.Fatalf("读取首批字节失败: %v", readErr)
+		}
+	case <-time.After(5 * time.Second):
+		unblock()
+		t.Fatalf("首播没有边下边发：上游卡在中途（仍有 %d 字节未发）时客户端读不到任何字节",
+			(chunks-stallAfter)*chunkBytes)
+	}
+	if string(first) != "m4a!" {
+		t.Fatalf("首批字节内容不对: %q", first)
+	}
+
+	// 放行上游，把剩余字节读完并确认缓存落盘完整。
+	unblock()
+	rest, readErr := io.ReadAll(resp.Body)
+	if readErr != nil {
+		t.Fatalf("读取剩余字节失败: %v", readErr)
+	}
+	if total := 4 + len(rest); total != chunks*chunkBytes {
+		t.Fatalf("交付字节 = %d，期望 %d", total, chunks*chunkBytes)
+	}
+	// 客户端读完最后一个字节时，服务端可能还在做 truncate + rename：
+	// 落盘完成是 handler 的收尾动作，给一个有限窗口等它。
+	deadline := time.Now().Add(2 * time.Second)
+	for !svc.Reader.HasCachedRemoteAudio(book.ID, upstream.URL+"/media") {
+		if time.Now().After(deadline) {
+			t.Fatalf("直通完成后应命中缓存")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}

internal/handler/reader_routes.go

@@ -13,6 +13,7 @@ import (
 	"path"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gin-gonic/gin"
@@ -131,16 +132,27 @@ func readerImportSourcesHandler(svc *service.Container) gin.HandlerFunc {
 func readerUpdateSourceHandler(svc *service.Container) gin.HandlerFunc {
 	var body struct {
 		Enabled *bool `json:"enabled"`
+		// MediaDirect 媒体直连开关：允许客户端自行下载音频/漫画图片字节。
+		// 可单独提交，也可与其他字段一起提交（只更新出现的字段）。
+		MediaDirect *bool `json:"media_direct"`
 	}
 	return func(c *gin.Context) {
-		if err := c.ShouldBindJSON(&body); err != nil || body.Enabled == nil {
-			c.JSON(http.StatusBadRequest, gin.H{"error": "enabled 字段必填"})
+		if err := c.ShouldBindJSON(&body); err != nil || (body.Enabled == nil && body.MediaDirect == nil) {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "enabled 或 media_direct 字段必填"})
 			return
 		}
 		userID := c.GetString(middleware.CtxUserID)
-		if err := svc.Reader.UpdateSourceEnabled(c.Request.Context(), userID, c.Param("id"), *body.Enabled); err != nil {
-			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
-			return
+		if body.Enabled != nil {
+			if err := svc.Reader.UpdateSourceEnabled(c.Request.Context(), userID, c.Param("id"), *body.Enabled); err != nil {
+				c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+				return
+			}
+		}
+		if body.MediaDirect != nil {
+			if err := svc.Reader.SetSourceMediaDirect(c.Request.Context(), userID, c.Param("id"), *body.MediaDirect); err != nil {
+				c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+				return
+			}
 		}
 		c.JSON(http.StatusOK, gin.H{"ok": true})
 	}
@@ -1004,8 +1016,61 @@ const (
 // readerAudioFetchGroup 同一音轨并发请求（播放器 + 预取 + 预热）只下载一次。
 var readerAudioFetchGroup singleflight.Group
 
-// handleCachedReaderAudio 音频直链的统一入口：先查本地缓存，未命中则整段拉取落盘，
-// 最后交给 http.ServeContent 处理 Range（拖动进度、续播都读本地文件）。
+// readerAudioStreamLocks 首播直通按缓存键串行化。
+//
+// 直通模式下两个并发请求会同时写同一个 .part 文件（播放器 + 预热器各自开一条
+// 连接很常见），交错写入必然把文件写坏。这里让后到者排队等前一个下完，
+// 拿到锁后再查一次缓存直接下发——语义与原来 singleflight 等价，只是领导者
+// 是在流式外发的，追随者不需要重复下载。
+var readerAudioStreamLocks = newKeyedMutex()
+
+// keyedMutex 按字符串键串行化，锁释放后回收条目，避免 map 无限增长。
+type keyedMutex struct {
+	mu    sync.Mutex
+	locks map[string]*keyedMutexEntry
+}
+
+type keyedMutexEntry struct {
+	mu   sync.Mutex
+	refs int
+}
+
+func newKeyedMutex() *keyedMutex {
+	return &keyedMutex{locks: map[string]*keyedMutexEntry{}}
+}
+
+// Lock 取得该键的独占锁，返回解锁函数（必须调用）。
+func (k *keyedMutex) Lock(key string) func() {
+	k.mu.Lock()
+	entry := k.locks[key]
+	if entry == nil {
+		entry = &keyedMutexEntry{}
+		k.locks[key] = entry
+	}
+	entry.refs++
+	k.mu.Unlock()
+
+	entry.mu.Lock()
+	return func() {
+		entry.mu.Unlock()
+		k.mu.Lock()
+		entry.refs--
+		if entry.refs == 0 {
+			delete(k.locks, key)
+		}
+		k.mu.Unlock()
+	}
+}
+
+// handleCachedReaderAudio 音频直链的统一入口。
+//
+// 三条路径，按代价从低到高：
+//  1. 命中完整缓存 → http.ServeContent 按 Range 下发（拖动、续播都读本地）；
+//  2. 冷启动首播 → 边下边发给客户端，同时落盘（DownloadRemoteAudioStreaming）。
+//     旧实现是「整段下完再下发」，慢链路上用户要对着转圈等整章 3–5MB，
+//     把「偶发失败」换成了「必然慢」；直通后几百毫秒就能出声，落盘照旧；
+//  3. 已有 .part 分片 → 走整段下载：分片意味着上次已经下到一半，同步续传
+//     剩余字节比重新直通更省流量，下完再一次性下发。
 func handleCachedReaderAudio(
 	c *gin.Context,
 	svc *service.Container,
@@ -1018,6 +1083,18 @@ func handleCachedReaderAudio(
 		serveCachedReaderAudio(c, f, info, rawURL)
 		return
 	}
+	// 播放器拖进度时会带 Range。已缓存的文件才谈得上 Range；未缓存时直通
+	// 只能从 0 发，否则客户端拿到的区间与它请求的对不上。
+	rangeHeader := strings.TrimSpace(c.GetHeader("Range"))
+	canStream := c.Request.Method == http.MethodGet &&
+		rangeHeader == "" &&
+		!svc.Reader.HasPartialRemoteAudio(bookID, rawURL)
+
+	if canStream {
+		handleStreamingReaderAudio(c, svc, book, bookID, rawURL, headers)
+		return
+	}
+
 	cachePath, err := ensureCachedRemoteAudio(c.Request.Context(), svc, book, bookID, rawURL, headers)
 	if err != nil {
 		if svc.Log != nil {
@@ -1040,6 +1117,97 @@ func handleCachedReaderAudio(
 	serveCachedReaderAudio(c, f, info, rawURL)
 }
 
+// handleStreamingReaderAudio 首播直通：一边落盘一边把字节写给客户端。
+//
+// 直接写 c.Writer 而不是先落盘再 ServeContent，是为了让第一个字节尽早到达。
+// 下载失败时客户端已经收到部分字节，只能中止响应（见下面的 panic），
+// 不能改成 502 —— 状态码早就发出去了。
+func handleStreamingReaderAudio(
+	c *gin.Context,
+	svc *service.Container,
+	book *model.ReaderBook,
+	bookID, rawURL string,
+	headers map[string]string,
+) {
+	key := svc.Reader.RemoteAudioCachePath(bookID, rawURL)
+	if key == "" {
+		c.String(http.StatusBadGateway, "音频缓存不可用")
+		return
+	}
+	unlock := readerAudioStreamLocks.Lock(key)
+	defer unlock()
+
+	// 排队期间前一个请求可能已经把整章下完：直接下发本地文件，不重复回源。
+	if f, info, ok := svc.Reader.OpenCachedRemoteAudio(bookID, rawURL); ok {
+		defer f.Close()
+		serveCachedReaderAudio(c, f, info, rawURL)
+		return
+	}
+
+	writer := c.Writer
+	// started 表示响应头是否已发出。没发出时失败还能改成 502；已经发出就只能
+	// 中止连接——状态码收不回来了。
+	started := false
+	_, err := svc.Reader.DownloadRemoteAudioStreaming(
+		c.Request.Context(), bookID, rawURL, readerAudioCacheAttempts,
+		readerAudioFetchFunc(c.Request.Context(), svc, book, bookID, rawURL, headers),
+		func(total int64, upstreamType string) {
+			contentType := readerAudioContentType(rawURL)
+			// 上游给了明确类型时优先用它（个别源把 m4a 标成 audio/x-m4a）。
+			if strings.HasPrefix(strings.ToLower(upstreamType), "audio/") {
+				contentType = upstreamType
+			}
+			writer.Header().Set("Content-Type", contentType)
+			writer.Header().Set("Cache-Control", "private, max-age=604800")
+			if total > 0 {
+				writer.Header().Set("Content-Length", strconv.FormatInt(total, 10))
+			}
+			writer.WriteHeader(http.StatusOK)
+			started = true
+			// 立刻把已写好的头刷给客户端：gin 的 ResponseWriter 有缓冲，
+			// 不 Flush 的话播放器要等到第一批音频字节才看到响应，白等一轮 RTT。
+			writer.Flush()
+		},
+		writer,
+	)
+	if err == nil {
+		return
+	}
+	if svc.Log != nil {
+		svc.Log.Warn("reader 音频直通失败", zap.String("book", bookID), zap.Error(err))
+	}
+	if !started {
+		// 一个字节都还没发出去（首拉就失败）：保持与旧实现一致的错误语义。
+		c.String(http.StatusBadGateway, "音频拉取失败: %s", err.Error())
+		return
+	}
+	// 响应头已经发出：只能中止连接，让客户端判定「传输不完整」并自己重试。
+	// 保留 .part 分片，下一次请求会走续传路径。
+	panic(http.ErrAbortHandler)
+}
+
+// readerAudioFetchFunc 把「带换票重试的媒体拉取」适配成下载器需要的 fetcher。
+func readerAudioFetchFunc(
+	ctx context.Context,
+	svc *service.Container,
+	book *model.ReaderBook,
+	bookID, rawURL string,
+	headers map[string]string,
+) reader.AudioRangeFetcher {
+	return func(rangeHeader string) (*http.Response, []byte, error) {
+		// 只有整段请求才需要嗅探响应体（判 HLS/错误页）；续传拿到的是 206
+		// 二进制分片，读了还得再补写一次，没必要。
+		out, fetchErr := fetchReaderMediaResilient(ctx, svc, book, rawURL, rangeHeader, headers, rangeHeader == "")
+		if fetchErr != nil {
+			if isReaderMediaNoRetry(fetchErr) {
+				return nil, nil, &reader.NoRetryError{Err: fetchErr}
+			}
+			return nil, nil, fetchErr
+		}
+		return out.resp, out.head, nil
+	}
+}
+
 // serveCachedReaderAudio 用 http.ServeContent 下发本地音轨（自带 Range / 206 语义）。
 func serveCachedReaderAudio(c *gin.Context, f *os.File, info os.FileInfo, rawURL string) {
 	c.Header("Content-Type", readerAudioContentType(rawURL))
@@ -1089,18 +1257,7 @@ func downloadReaderAudioToCache(
 	headers map[string]string,
 ) error {
 	_, err := svc.Reader.DownloadRemoteAudio(ctx, bookID, rawURL, readerAudioCacheAttempts,
-		func(rangeHeader string) (*http.Response, []byte, error) {
-			// 只有整段请求才需要嗅探响应体（判 HLS/错误页）；续传拿到的是 206
-			// 二进制分片，读了还得再补写一次，没必要。
-			out, fetchErr := fetchReaderMediaResilient(ctx, svc, book, rawURL, rangeHeader, headers, rangeHeader == "")
-			if fetchErr != nil {
-				if isReaderMediaNoRetry(fetchErr) {
-					return nil, nil, &reader.NoRetryError{Err: fetchErr}
-				}
-				return nil, nil, fetchErr
-			}
-			return out.resp, out.head, nil
-		})
+		readerAudioFetchFunc(ctx, svc, book, bookID, rawURL, headers))
 	return err
 }
 
@@ -1529,7 +1686,28 @@ func readerLocalAudioHandler(svc *service.Container) gin.HandlerFunc {
 	}
 }
 
+// readerContentOptions 从查询参数解析正文下发的可选项。
+//
+// 目前只有 `media` 一个维度：
+//   - 不传（网页端）：媒体全部走服务端签名代理，与历史行为一致；
+//   - `media=direct`：书源开启直连时保留上游地址，由客户端自取；
+//   - `media=proxy`：本次强制代理，给客户端直连失败时的退路。
+func readerContentOptions(c *gin.Context) reader.ContentOptions {
+	switch strings.ToLower(strings.TrimSpace(c.Query("media"))) {
+	case "direct":
+		return reader.ContentOptions{MediaDirect: true}
+	case "proxy":
+		return reader.ContentOptions{ForceProxy: true}
+	default:
+		return reader.ContentOptions{}
+	}
+}
+
 // readerBookContentHandler 书架维度正文（服务端应用替换净化规则）。
+//
+// media=direct 是客户端（Android App）专用的可选能力：媒体地址保留上游原始
+// 地址并由客户端自行下载，服务端只下发所需请求头。不传该参数（网页端）时
+// 行为与历史版本完全一致。
 func readerBookContentHandler(svc *service.Container) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		chapter, err := strconv.Atoi(c.DefaultQuery("chapter", "0"))
@@ -1538,7 +1716,8 @@ func readerBookContentHandler(svc *service.Container) gin.HandlerFunc {
 			return
 		}
 		userID := c.GetString(middleware.CtxUserID)
-		content, err := svc.Reader.GetContentForBook(c.Request.Context(), userID, c.Param("id"), chapter)
+		opts := readerContentOptions(c)
+		content, err := svc.Reader.GetContentForBook(c.Request.Context(), userID, c.Param("id"), chapter, opts)
 		if err != nil {
 			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 			return
@@ -1551,13 +1730,14 @@ func readerBookContentHandler(svc *service.Container) gin.HandlerFunc {
 //
 // 请求 {chapter_indexes:[...], apply_replace:bool}：单章失败不中断整批，
 // 返回每章内容与 hit/miss/failed 统计，供前端窗口预取与后续离线缓存。
+// 媒体通道同样按 media=direct 查询参数选择，默认（网页端）走服务端代理。
 func readerBookContentBatchHandler(svc *service.Container) gin.HandlerFunc {
-	var body struct {
-		ChapterIndexes []int `json:"chapter_indexes" binding:"required"`
-		// ApplyReplace 是否在服务端应用用户替换净化规则（默认 false，与单章接口一致）。
-		ApplyReplace bool `json:"apply_replace"`
-	}
 	return func(c *gin.Context) {
+		var body struct {
+			ChapterIndexes []int `json:"chapter_indexes" binding:"required"`
+			// ApplyReplace 是否在服务端应用用户替换净化规则（默认 false，与单章接口一致）。
+			ApplyReplace bool `json:"apply_replace"`
+		}
 		if err := c.ShouldBindJSON(&body); err != nil {
 			c.JSON(http.StatusBadRequest, gin.H{"error": "chapter_indexes 不能为空"})
 			return
@@ -1567,7 +1747,7 @@ func readerBookContentBatchHandler(svc *service.Container) gin.HandlerFunc {
 			return
 		}
 		userID := c.GetString(middleware.CtxUserID)
-		result, err := svc.Reader.GetContentBatch(c.Request.Context(), userID, c.Param("id"), body.ChapterIndexes, body.ApplyReplace)
+		result, err := svc.Reader.GetContentBatch(c.Request.Context(), userID, c.Param("id"), body.ChapterIndexes, body.ApplyReplace, readerContentOptions(c))
 		if err != nil {
 			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 			return