diff --git a/internal/config/types.go b/internal/config/types.go index 252703c..052d882 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -123,7 +123,8 @@ type CacheConfig struct { CacheDir string `mapstructure:"cache_dir"` // ImagesMaxSizeMB 是整个磁盘缓存的总容量上限(MB),不是 images 子目录 // 的配额。参与统计的池为 hls、images、library-covers;超限时按该顺序淘汰 - // 最旧文件。reader-audio 由自身 1GB 配额在写入时独立淘汰,不参与总容量。 + // 最旧文件。reader-audio 不在其中:它按 reader.audioMediaMaxBytes 的固定 + // 配额(当前 1GiB)在写入时独立淘汰,不参与这里的总量。 // 0 表示不限制。 ImagesMaxSizeMB int `mapstructure:"images_max_size_mb"` // ImagesOriginalsMaxSizeMB 限制 cache/images 下原图子配额(MB)。原图只是 diff --git a/internal/handler/reader_media_test.go b/internal/handler/reader_media_test.go index 29b0ad1..3633cd1 100644 --- a/internal/handler/reader_media_test.go +++ b/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) + } +} diff --git a/internal/handler/reader_routes.go b/internal/handler/reader_routes.go index d5800c9..d8ab5b6 100644 --- a/internal/handler/reader_routes.go +++ b/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 diff --git a/internal/model/reader.go b/internal/model/reader.go index 358c49e..3835c67 100644 --- a/internal/model/reader.go +++ b/internal/model/reader.go @@ -38,6 +38,12 @@ type ReaderBookSource struct { // NeedsBrowser 是否依赖 WebView/无头浏览器(webView 选项或 webjs 规则)。 // 服务端没有浏览器,这类源注定不可用,列表接口按需计算让前端能提前提示。 NeedsBrowser bool `gorm:"-" json:"needs_browser"` + // MediaDirect 允许客户端直连上游取实体媒体(音频/漫画图片),服务端只负责 + // 解析地址与下发所需请求头。 + // + // 只在客户端显式请求直连时生效(见 ContentOptions):网页端不声明该能力, + // 行为与开启此字段前完全一致。 + MediaDirect bool `gorm:"default:false" json:"media_direct"` } // ReaderSourceState 书源会话状态:对应 legado 中按书源 key 存储的 diff --git a/internal/service/cache_cleanup.go b/internal/service/cache_cleanup.go index e2cfa75..772020e 100644 --- a/internal/service/cache_cleanup.go +++ b/internal/service/cache_cleanup.go @@ -386,7 +386,8 @@ type cacheEvictionEntry struct { // cacheEvictionPoolOrder 是全局容量清理的固定优先级。 // 顺序即“先清谁”:超限时前面的池会被整池清空,才会轮到后面的池。 -// reader-audio 不在其中:它由自身 1GB 配额在写入时独立淘汰。 +// reader-audio 不在其中:它由自身固定配额(当前 1GiB,见 reader 包的 +// audioMediaMaxBytes)在写入时独立淘汰。 var cacheEvictionPoolOrder = []string{ "hls", "images", @@ -496,9 +497,10 @@ func removeEmptyDirsUnder(root string) { // PruneCache 按全局容量池清理磁盘缓存。 // // 总量口径是 hls、images、library-covers 三个池之和;reader-audio 及正文缓存、 -// 书源文件缓存都不参与这里的总量计算——reader-audio 由自身 1GB 配额在写入时 -// 淘汰,正文缓存和书源文件缓存另有自己的 TTL / 配额。超过 totalBytes 时严格 -// 按 hls → images → library-covers 的顺序淘汰最旧文件,直到回落到上限的 80%。 +// 书源文件缓存都不参与这里的总量计算——reader-audio 由自身固定配额(当前 +// 1GiB)在写入时淘汰,正文缓存和书源文件缓存另有自己的 TTL / 配额。 +// 超过 totalBytes 时严格按 hls → images → library-covers 的顺序淘汰最旧文件, +// 直到回落到上限的 80%。 // originalsMaxBytes 是 images 原图池的独立子配额,在总量淘汰之后单独执行 // (即使总量没超也会把原图压回配额)。 func PruneCache(cacheDir string, totalBytes, originalsMaxBytes int64) (CacheCleanupResult, error) { diff --git a/internal/service/reader/audio_media_cache.go b/internal/service/reader/audio_media_cache.go index 7c2110b..c0c8d1e 100644 --- a/internal/service/reader/audio_media_cache.go +++ b/internal/service/reader/audio_media_cache.go @@ -22,14 +22,18 @@ import ( // // 背景:听书源的音频直链签在腾讯音乐 CDN 上,跨境链路会间歇性返回 401 // 或直接 RST 连接。旧行为是每次请求(首播、拖动进度、播放器重试)都重新 -// 回源,一次抖动就把错误交给播放器。这里按「书 + 媒体对象路径」把整段音频 -// 落盘(忽略 ?vkey=... 这类会轮换的签名参数),命中后由 http.ServeContent +// 回源,一次抖动就把错误交给播放器。这里按「书 + 媒体对象」把整段音频 +// 落盘(丢掉 ?vkey=... 这类会轮换的签名参数),命中后由 http.ServeContent // 按 Range 下发:首播成功之后不再依赖 CDN,拖动进度也是本地读取。 const ( - audioMediaDirName = "reader-audio" - audioMediaTTL = 7 * 24 * time.Hour - audioMediaMaxBytes = 2 << 30 // 2GiB:超出后按 mtime 淘汰最旧的 + audioMediaDirName = "reader-audio" + audioMediaTTL = 7 * 24 * time.Hour + // audioMediaMaxBytes 音频缓存的总容量上限。超限时按 mtime 淘汰最旧的。 + // + // 这个池不参与 cache_cleanup 的全局总量统计(hls/images/library-covers + // 三池),而是在每次下载成功后自行淘汰,见 pruneAudioMediaCache。 + audioMediaMaxBytes = 1 << 30 // 1GiB audioMediaPrunePeriod = 10 * time.Minute audioMediaPruneKeepPct = 90 @@ -68,16 +72,48 @@ func (s *ReaderService) audioMediaDir() string { return filepath.Join(base, audioMediaDirName) } -// audioMediaCacheKey 缓存键:书 + scheme://host/path。 +// audioMediaVolatileParams 是换票时会变、但与「这是哪段音频」无关的 query 参数。 +// 缓存键必须忽略它们,否则每次换票都打不中缓存。 // -// 必须忽略 query:同一段音频每次换票都会得到新的 vkey/uid/ct,只按完整 URL -// 取哈希的话缓存永远打不中。 +// 不在这张表里的参数一律保留:有些书源的直链形如 +// https://host/audio?id=12345,路径固定、靠 query 区分章节,把 query 整个丢掉 +// 会让第二集命中第一集的缓存,直接播错集。 +var audioMediaVolatileParams = map[string]bool{ + "vkey": true, "uid": true, "ct": true, "t": true, "ts": true, + "token": true, "sign": true, "signature": true, "sig": true, + "expires": true, "expire": true, "expiry": true, "auth": true, + "nonce": true, "rand": true, "rnd": true, "r": true, "guid": true, + "ftime": true, +} + +// audioMediaCacheKey 缓存键:书 + 媒体对象身份。 +// +// 身份 = scheme://host/path + 非签名类 query。签名/时效参数(vkey、token、 +// 时间戳等)被剔除,保证换票后仍命中同一条缓存;其余参数保留,避免 +// 「路径相同、只靠 query 区分」的书源串集。 func audioMediaCacheKey(bookID, mediaURL string) string { u, err := url.Parse(strings.TrimSpace(mediaURL)) if err != nil || u.Host == "" { return "" } ident := strings.ToLower(u.Scheme) + "://" + strings.ToLower(u.Host) + u.EscapedPath() + if u.RawQuery != "" { + if q, parseErr := url.ParseQuery(u.RawQuery); parseErr == nil { + var kept []string + for k, vs := range q { + if audioMediaVolatileParams[strings.ToLower(k)] { + continue + } + for _, v := range vs { + kept = append(kept, k+"="+v) + } + } + if len(kept) > 0 { + sort.Strings(kept) + ident += "?" + strings.Join(kept, "&") + } + } + } sum := sha256.Sum256([]byte(bookID + "|" + ident)) return hex.EncodeToString(sum[:16]) } @@ -121,6 +157,23 @@ func (s *ReaderService) HasCachedRemoteAudio(bookID, mediaURL string) bool { return ok } +// HasPartialRemoteAudio 是否有可续传的未完成分片(.part)。 +// +// 调用方据此决定首播策略:有分片说明上一次已经下到一半(用户点过重试、 +// 或播放器自己重连过),此时「下载完再下发」能透明续上剩余字节; +// 没有分片才是真正的冷启动,边下边发换更低的 TTFB。 +func (s *ReaderService) HasPartialRemoteAudio(bookID, mediaURL string) bool { + path := s.RemoteAudioCachePath(bookID, mediaURL) + if path == "" { + return false + } + info, err := os.Stat(path + audioPartialSuffix) + if err != nil || !info.Mode().IsRegular() || info.Size() == 0 { + return false + } + return info.Size() <= audioMediaObjectMaxBytes +} + // OpenCachedRemoteAudio 打开已缓存的音轨(命中返回打开的文件,调用方负责关闭)。 // 过期的缓存顺手删掉,未过期则刷新 mtime 供 LRU 淘汰。 func (s *ReaderService) OpenCachedRemoteAudio(bookID, mediaURL string) (*os.File, os.FileInfo, bool) { @@ -166,6 +219,23 @@ func (e *NoRetryError) Unwrap() error { return e.Err } // 调用方负责在内部完成换票;这里只看状态码和字节。 type AudioRangeFetcher func(rangeHeader string) (resp *http.Response, head []byte, err error) +// AudioStreamStart 首播直通开始的回调:调用方在这里写响应头与状态码。 +// total 是上游声明的总长度(分块传输时为 -1),contentType 是上游给的类型。 +type AudioStreamStart func(total int64, contentType string) + +// errAudioStreamCannotRestart 直通模式下遇到必须「从第 0 字节重来」的情况。 +// +// 直通时客户端已经收到了前缀字节,不可能让它把已收到的丢掉重听,所以这类 +// 分支只能中止这次响应(让客户端判成不完整),并把已下载的分片留给下次续传。 +var errAudioStreamCannotRestart = errors.New("直通中无法从零重新下载") + +// audioSinkError 标记「写客户端失败」(通常是播放器关掉了连接)。 +// 与写盘失败区分开:写盘失败说明本地有问题,写客户端失败只需停止下载。 +type audioSinkError struct{ err error } + +func (e *audioSinkError) Error() string { return "写入客户端失败: " + e.err.Error() } +func (e *audioSinkError) Unwrap() error { return e.err } + // DownloadRemoteAudio 带断点续传地把整段音轨写进缓存,返回最终文件路径。 // // 为什么需要续传:腾讯 CDN 的跨境边缘常在传输中途 RST,旧实现每次重试都从 @@ -181,6 +251,40 @@ func (s *ReaderService) DownloadRemoteAudio( bookID, mediaURL string, attempts int, fetch AudioRangeFetcher, +) (string, error) { + return s.downloadRemoteAudio(ctx, bookID, mediaURL, attempts, fetch, nil, nil) +} + +// DownloadRemoteAudioStreaming 首播直通:一边落盘一边把字节写给 sink。 +// +// 与 DownloadRemoteAudio 的区别只有一处——不再等整段下完才返回,第一个字节 +// 到达就通过 onStart 写好响应头并开始外发,慢链路上的首播不必先等 3–5MB。 +// +// 代价是失败语义变严:客户端已经收到前缀字节,任何「必须从 0 重来」的情况 +// (上游忽略 Range 回 200、分片区间对不上、416)都只能中止这次响应, +// 不能像普通模式那样静默重下。已写入的分片仍然保留,下次可续传。 +func (s *ReaderService) DownloadRemoteAudioStreaming( + ctx context.Context, + bookID, mediaURL string, + attempts int, + fetch AudioRangeFetcher, + onStart AudioStreamStart, + sink io.Writer, +) (string, error) { + if sink == nil { + return "", errors.New("直通下载未提供输出目标") + } + return s.downloadRemoteAudio(ctx, bookID, mediaURL, attempts, fetch, onStart, sink) +} + +// downloadRemoteAudio 是两种下载模式的共同实现。sink 非空即首播直通模式。 +func (s *ReaderService) downloadRemoteAudio( + ctx context.Context, + bookID, mediaURL string, + attempts int, + fetch AudioRangeFetcher, + onStart AudioStreamStart, + sink io.Writer, ) (string, error) { path := s.RemoteAudioCachePath(bookID, mediaURL) if path == "" { @@ -209,6 +313,8 @@ func (s *ReaderService) DownloadRemoteAudio( if err != nil { return "", fmt.Errorf("创建音频缓存文件失败: %w", err) } + // onStart 只应触发一次:直通模式下响应头只能写一次。 + started := false var lastErr error for attempt := 1; attempt <= attempts; attempt++ { @@ -250,6 +356,12 @@ func (s *ReaderService) DownloadRemoteAudio( start, total, ok := parseContentRange(resp.Header.Get("Content-Range")) if !ok || offset == 0 || start != offset { drainAndClose(resp.Body) + if sink != nil && offset > 0 { + // 已经发给客户端 offset 字节:重来会让客户端收到重复内容。 + // 分片保持原样,下次请求从这里续。 + _ = file.Close() + return "", errAudioStreamCannotRestart + } offset = 0 _ = file.Truncate(0) lastErr = fmt.Errorf("音源返回的分片区间不可用(%q)", resp.Header.Get("Content-Range")) @@ -259,9 +371,14 @@ func (s *ReaderService) DownloadRemoteAudio( case http.StatusOK: // 上游忽略了 Range(或者本来就是整段请求):只能从第 0 字节写。 if offset > 0 { + drainAndClose(resp.Body) + if sink != nil { + // 同上:已发出去的字节不能撤回,分片留着供下次续传。 + _ = file.Close() + return "", errAudioStreamCannotRestart + } offset = 0 if truncErr := file.Truncate(0); truncErr != nil { - drainAndClose(resp.Body) lastErr = fmt.Errorf("重置音频缓存失败: %w", truncErr) continue } @@ -270,6 +387,10 @@ func (s *ReaderService) DownloadRemoteAudio( case http.StatusRequestedRangeNotSatisfiable: // 分片比上游对象还长(对象换过):丢掉分片重来。 drainAndClose(resp.Body) + if sink != nil && offset > 0 { + _ = file.Close() + return "", errAudioStreamCannotRestart + } offset = 0 _ = file.Truncate(0) lastErr = fmt.Errorf("音源分片已失效(%d)", resp.StatusCode) @@ -280,12 +401,27 @@ func (s *ReaderService) DownloadRemoteAudio( continue } + if sink != nil && !started { + // 直通模式下要在这里就把状态码与 Content-Length 写出去,客户端才能 + // 边收边播。写头的失败(客户端已断开)当作停止下载处理。 + if onStart != nil { + onStart(expected, resp.Header.Get("Content-Type")) + } + started = true + } + if _, seekErr := file.Seek(offset, io.SeekStart); seekErr != nil { drainAndClose(resp.Body) lastErr = fmt.Errorf("音频缓存定位失败: %w", seekErr) continue } - written, writeErr := writeAudioChunk(file, head, resp.Body, audioMediaObjectMaxBytes-offset) + var written int64 + var writeErr error + if sink != nil { + written, writeErr = writeAudioChunkStream(file, sink, head, resp.Body, audioMediaObjectMaxBytes-offset) + } else { + written, writeErr = writeAudioChunk(file, head, resp.Body, audioMediaObjectMaxBytes-offset) + } _ = resp.Body.Close() offset += written if errors.Is(writeErr, errAudioObjectTooLarge) { @@ -293,6 +429,12 @@ func (s *ReaderService) DownloadRemoteAudio( _ = os.Remove(partPath) return "", fmt.Errorf("已重试 %d 次: %w", attempts, writeErr) } + var sinkErr *audioSinkError + if errors.As(writeErr, &sinkErr) { + // 客户端不要了:停止下载,已写入的分片留着供下次续传。 + _ = file.Close() + return "", sinkErr + } if writeErr != nil { // 已写入的字节保留在分片里,下一轮从新的断点续传。 lastErr = writeErr @@ -355,6 +497,64 @@ func writeAudioChunk(w io.Writer, head []byte, body io.Reader, limit int64) (int return total, nil } +// writeAudioChunkStream 同 writeAudioChunk,但同时把字节写给客户端(首播直通)。 +// +// 写盘失败要让整个下载停下来(本地状态不可信,继续发字节等于给客户端一份 +// 服务端存不下的内容);写客户端失败只说明接收方不要了,用 audioSinkError +// 标出来让上层保留分片、安静收尾。两者必须区分,否则播放器一关就把分片删了。 +func writeAudioChunkStream(w io.Writer, sink io.Writer, head []byte, body io.Reader, limit int64) (int64, error) { + var total int64 + writeBoth := func(p []byte) error { + n, err := w.Write(p) + total += int64(n) + if err != nil { + return err + } + if n > 0 { + if _, sinkErr := sink.Write(p[:n]); sinkErr != nil { + return &audioSinkError{err: sinkErr} + } + } + return nil + } + if len(head) > 0 { + if err := writeBoth(head); err != nil { + return total, err + } + } + if limit <= 0 { + return total, errAudioObjectTooLarge + } + buf := make([]byte, 32<<10) + remaining := limit - total + 1 + for { + if remaining <= 0 { + return total, errAudioObjectTooLarge + } + chunk := buf + if int64(len(chunk)) > remaining { + chunk = chunk[:remaining] + } + n, readErr := body.Read(chunk) + if n > 0 { + if err := writeBoth(chunk[:n]); err != nil { + return total, err + } + remaining -= int64(n) + } + if readErr == io.EOF { + break + } + if readErr != nil { + return total, readErr + } + } + if total > limit { + return total, errAudioObjectTooLarge + } + return total, nil +} + // parseContentRange 解析 "bytes 100-199/1234" 的起点与总长度。 // 总长度未知("*")时 total 为 -1。 func parseContentRange(v string) (start, total int64, ok bool) { diff --git a/internal/service/reader/audio_media_cache_test.go b/internal/service/reader/audio_media_cache_test.go index 1d21682..7d83a25 100644 --- a/internal/service/reader/audio_media_cache_test.go +++ b/internal/service/reader/audio_media_cache_test.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "os" + "path/filepath" "strings" "testing" @@ -35,8 +36,8 @@ func TestIsAudioMediaURL(t *testing.T) { } } -// TestAudioMediaCacheKeyIgnoresQuery 换票会改写 uid/vkey/ct,缓存键必须只看 -// scheme + host + path,否则每次换票都会重新下载一遍。 +// TestAudioMediaCacheKeyIgnoresQuery 换票会改写 uid/vkey/ct,缓存键必须忽略 +// 这些签名/时效参数,否则每次换票都会重新下载一遍。 func TestAudioMediaCacheKeyIgnoresQuery(t *testing.T) { a := audioMediaCacheKey("book-1", "https://cdn.example.com/long/a.m4a?uid=1&vkey=aaa") b := audioMediaCacheKey("book-1", "https://cdn.example.com/long/a.m4a?uid=2&vkey=bbb&ct=3") @@ -51,6 +52,30 @@ func TestAudioMediaCacheKeyIgnoresQuery(t *testing.T) { } } +// TestAudioMediaCacheKeyKeepsIdentifyingQuery 只靠 query 区分章节的书源 +// (路径固定、如 /audio?id=12345)不能被合并成同一个缓存对象,否则第二集 +// 会命中第一集的缓存直接播错内容。 +func TestAudioMediaCacheKeyKeepsIdentifyingQuery(t *testing.T) { + ep1 := audioMediaCacheKey("book-1", "https://host/audio?id=12345&vkey=aaa") + ep2 := audioMediaCacheKey("book-1", "https://host/audio?id=67890&vkey=bbb") + if ep1 == "" || ep2 == "" { + t.Fatal("缓存键不应为空") + } + if ep1 == ep2 { + t.Fatalf("不同章节(query 不同)的缓存键不应相同,否则会串集") + } +} + +// TestAudioMediaCacheKeyIgnoresQueryOrder 非签名参数的顺序不应影响缓存键。 +func TestAudioMediaCacheKeyIgnoresQueryOrder(t *testing.T) { + a := audioMediaCacheKey("book-1", "https://host/a.m4a?id=7&chapter=3&vkey=x") + b := audioMediaCacheKey("book-1", "https://host/a.m4a?chapter=3&id=7&vkey=y") + if a == "" || a != b { + t.Fatalf("参数顺序不应影响缓存键: %q vs %q", a, b) + } +} + + // ─── 断点续传下载器 ────────────────────────────────────────────────────── func TestParseContentRange(t *testing.T) { @@ -267,3 +292,143 @@ func TestDownloadRemoteAudioStopsOnNoRetry(t *testing.T) { t.Fatalf("不可重试的失败只应请求 1 次,实际 %d 次", calls) } } + +// ─── 首播直通(边下边发) ──────────────────────────────────────────────── + +// TestDownloadRemoteAudioStreamingForwardsWhileDownloading 直通模式:字节必须 +// 在下载过程中就到达 sink,而不是等整段下完才一次性写出去(首播 TTFB)。 +func TestDownloadRemoteAudioStreamingForwardsWhileDownloading(t *testing.T) { + svc := newAudioDownloadTestService(t) + const mediaURL = "https://cdn.example.com/long/stream.m4a? vkey=1" + payload := bytes.Repeat([]byte("stream!"), 1000) + + var sink bytes.Buffer + var startedTotal int64 + // 用「读到一半时 sink 里已经有数据」来证明确实在边下边发。 + var midLength int + path, err := svc.DownloadRemoteAudioStreaming(t.Context(), "book-s", mediaURL, 1, + func(rangeHeader string) (*http.Response, []byte, error) { + if rangeHeader != "" { + t.Fatalf("首播直通不应带 Range,实际 %q", rangeHeader) + } + return okResponse(payload), nil, nil + }, + func(total int64, contentType string) { + startedTotal = total + // 头部回调必须早于任何字节下发。 + if sink.Len() != 0 { + t.Errorf("写响应头时不应已有字节写出") + } + }, + writerFunc(func(p []byte) (int, error) { + if midLength == 0 && sink.Len() == 0 && len(p) > 0 { + midLength = sink.Len() + } + return sink.Write(p) + }), + ) + if err != nil { + t.Fatalf("直通下载失败: %v", err) + } + if startedTotal != int64(len(payload)) { + t.Fatalf("头部回调应拿到上游总长 %d,实际 %d", len(payload), startedTotal) + } + if !bytes.Equal(sink.Bytes(), payload) { + t.Fatalf("下发的字节不完整: %d 字节,期望 %d", sink.Len(), len(payload)) + } + // 下载完成后缓存文件必须完整可命中。 + got, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatalf("读缓存文件失败: %v", readErr) + } + if !bytes.Equal(got, payload) { + t.Fatalf("落盘内容不完整: %d 字节", len(got)) + } + if !svc.HasCachedRemoteAudio("book-s", mediaURL) { + t.Fatal("直通下完后应当命中缓存") + } +} + +// TestDownloadRemoteAudioStreamingStopsWhenClientGoesAway 客户端中途断开时 +// 停止下载,但保留分片供下次续传(不能把已下的字节白扔掉)。 +func TestDownloadRemoteAudioStreamingStopsWhenClientGoesAway(t *testing.T) { + svc := newAudioDownloadTestService(t) + const mediaURL = "https://cdn.example.com/long/gone.m4a" + payload := bytes.Repeat([]byte("abcdefgh"), 2000) + clientGone := errors.New("client closed connection") + + var sink bytes.Buffer + _, err := svc.DownloadRemoteAudioStreaming(t.Context(), "book-g", mediaURL, 1, + func(rangeHeader string) (*http.Response, []byte, error) { + return okResponse(payload), nil, nil + }, + nil, + writerFunc(func(p []byte) (int, error) { + // 收下前 1000 字节后模拟播放器断开。 + if sink.Len() >= 1000 { + return 0, clientGone + } + room := 1000 - sink.Len() + if len(p) <= room { + return sink.Write(p) + } + n, _ := sink.Write(p[:room]) + return n, clientGone + }), + ) + if err == nil { + t.Fatal("客户端断开时应当停止下载") + } + var sinkErr *audioSinkError + if !errors.As(err, &sinkErr) { + t.Fatalf("应标记为写客户端失败(便于上层区分),实际 %v", err) + } + // 分片要留着:这是续传的原料。 + path := svc.RemoteAudioCachePath("book-g", mediaURL) + if path == "" { + t.Fatal("缓存路径不应为空") + } + if info, statErr := os.Stat(path + audioPartialSuffix); statErr != nil || info.Size() == 0 { + t.Fatalf("断流后应保留 .part 分片,实际 %v", statErr) + } + if svc.HasCachedRemoteAudio("book-g", mediaURL) { + t.Fatal("中断的下载不能留下完整缓存(否则会下发截断音频)") + } +} + +// TestDownloadRemoteAudioStreamingAbortsOnRestart 直通模式下上游忽略 Range 回 +// 200(需要从 0 重来)时必须中止,而不是把重复字节拼进客户端已收到的流里。 +func TestDownloadRemoteAudioStreamingAbortsOnRestart(t *testing.T) { + svc := newAudioDownloadTestService(t) + const mediaURL = "https://cdn.example.com/long/restart.m4a" + payload := bytes.Repeat([]byte("zz"), 100) + // 先留一个分片,让第二轮带 Range。 + partPath := svc.RemoteAudioCachePath("book-r", mediaURL) + audioPartialSuffix + if err := os.MkdirAll(filepath.Dir(partPath), 0o750); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(partPath, payload[:50], 0o640); err != nil { + t.Fatal(err) + } + + var sink bytes.Buffer + _, err := svc.DownloadRemoteAudioStreaming(t.Context(), "book-r", mediaURL, 1, + func(rangeHeader string) (*http.Response, []byte, error) { + // 无视 Range,整段回 200 → 直通无法重来。 + return okResponse(payload), nil, nil + }, + nil, + &sink, + ) + if !errors.Is(err, errAudioStreamCannotRestart) { + t.Fatalf("应返回不可重来错误,实际 %v", err) + } + if sink.Len() != 0 { + t.Fatalf("中止前不应向客户端写出字节,实际 %d", sink.Len()) + } +} + +// writerFunc 把函数适配成 io.Writer。 +type writerFunc func(p []byte) (int, error) + +func (f writerFunc) Write(p []byte) (int, error) { return f(p) } diff --git a/internal/service/reader/book_type_fallback_test.go b/internal/service/reader/book_type_fallback_test.go index 09e8c22..dc76e18 100644 --- a/internal/service/reader/book_type_fallback_test.go +++ b/internal/service/reader/book_type_fallback_test.go @@ -58,7 +58,7 @@ func TestAudioBookTypeFallbackBySource(t *testing.T) { t.Fatalf("保存目录失败: %v", err) } - out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("取正文失败: %v", err) } @@ -125,7 +125,7 @@ func TestDeclaredTextTypeWinsOverAudioSource(t *testing.T) { t.Fatalf("保存目录失败: %v", err) } - out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("取正文失败: %v", err) } diff --git a/internal/service/reader/comic_image_test.go b/internal/service/reader/comic_image_test.go index 6d14609..7459e8a 100644 --- a/internal/service/reader/comic_image_test.go +++ b/internal/service/reader/comic_image_test.go @@ -75,7 +75,7 @@ func TestGetContentForBookExtractsImageSrc(t *testing.T) { } _ = sourceID - out, err := svc.GetContentForBook(ctx, "u1", book.ID, 0) + out, err := svc.GetContentForBook(ctx, "u1", book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("取正文失败: %v", err) } diff --git a/internal/service/reader/comment_test.go b/internal/service/reader/comment_test.go index bed9534..c090a3f 100644 --- a/internal/service/reader/comment_test.go +++ b/internal/service/reader/comment_test.go @@ -207,7 +207,7 @@ func TestGetContentForBookExtractsComments(t *testing.T) { t.Fatalf("写入章节失败: %v", err) } - out, err := svc.GetContentForBook(ctx, "u1", book.ID, 0) + out, err := svc.GetContentForBook(ctx, "u1", book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("取正文失败: %v", err) } @@ -282,7 +282,7 @@ func TestGetContentForBookKeepsTextWithStandaloneImage(t *testing.T) { `` svc, bookID := setupTextChapterBook(t, content) - out, err := svc.GetContentForBook(t.Context(), "u1", bookID, 0) + out, err := svc.GetContentForBook(t.Context(), "u1", bookID, 0, ContentOptions{}) if err != nil { t.Fatalf("取正文失败: %v", err) } @@ -310,7 +310,7 @@ func TestGetContentForBookStripsBlockTagsKeepsText(t *testing.T) { `` svc, bookID := setupTextChapterBook(t, content) - out, err := svc.GetContentForBook(t.Context(), "u1", bookID, 0) + out, err := svc.GetContentForBook(t.Context(), "u1", bookID, 0, ContentOptions{}) if err != nil { t.Fatalf("取正文失败: %v", err) } diff --git a/internal/service/reader/media_direct_test.go b/internal/service/reader/media_direct_test.go new file mode 100644 index 0000000..f56c715 --- /dev/null +++ b/internal/service/reader/media_direct_test.go @@ -0,0 +1,168 @@ +package reader + +import ( + "strings" + "testing" +) + +// 本文件:媒体直连(客户端自己取音频/漫画图片字节)的回归测试。 +// +// 契约: +// - 默认(ContentOptions 零值)必须与历史行为完全一致:媒体地址一律改写成 +// 服务端签名代理,响应里不出现直连标记与请求头 —— 网页端依赖这一点。 +// - 客户端显式请求直连 + 书源开启开关时:保留上游原始地址,并下发取媒体 +// 所需的请求头。 +// - 客户端请求直连但书源没开开关时:退回代理,不能把未开启的源推到客户端直连。 + +// TestMediaDirectDefaultsToProxy 不传直连选项时行为不变(网页端契约)。 +func TestMediaDirectDefaultsToProxy(t *testing.T) { + svc, _, book := prepareTicketBook(t) + ctx := t.Context() + + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) + if err != nil { + t.Fatalf("读取失败: %v", err) + } + if len(out.Tracks) == 0 { + t.Fatal("应至少有一条音轨") + } + if !strings.HasPrefix(out.Tracks[0], "/api/reader/media?") { + t.Fatalf("默认应改写为服务端代理,实际 %q", out.Tracks[0]) + } + if out.MediaDirect { + t.Fatal("默认不应标记为直连") + } + if len(out.MediaHeaders) != 0 { + t.Fatalf("代理模式不应下发取媒体请求头,实际 %v", out.MediaHeaders) + } +} + +// TestMediaDirectRequiresSourceSwitch 书源没开开关时,即使客户端请求直连也只给代理。 +func TestMediaDirectRequiresSourceSwitch(t *testing.T) { + svc, srv, book := prepareTicketBook(t) + ctx := t.Context() + + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{MediaDirect: true}) + if err != nil { + t.Fatalf("读取失败: %v", err) + } + if out.MediaDirect { + t.Fatal("书源未开启直连,不应标记为直连") + } + if !strings.HasPrefix(out.Tracks[0], "/api/reader/media?") { + t.Fatalf("书源未开启直连时仍应走代理,实际 %q", out.Tracks[0]) + } + _ = srv +} + +// TestMediaDirectKeepsUpstreamURL 开启开关后:地址是上游原始地址,并下发请求头。 +func TestMediaDirectKeepsUpstreamURL(t *testing.T) { + svc, srv, book := prepareTicketBook(t) + ctx := t.Context() + + if err := svc.SetSourceMediaDirect(ctx, readerTestUserID, sourceIDOf(t, svc, book.Origin), true); err != nil { + t.Fatalf("开启直连失败: %v", err) + } + + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{MediaDirect: true}) + if err != nil { + t.Fatalf("读取失败: %v", err) + } + if !out.MediaDirect { + t.Fatal("书源已开启直连,响应应标记 media_direct") + } + if len(out.Tracks) == 0 { + t.Fatal("应至少有一条音轨") + } + track := out.Tracks[0] + if strings.HasPrefix(track, "/api/") { + t.Fatalf("直连模式不该返回服务端代理地址,实际 %q", track) + } + if !strings.HasPrefix(track, srv.URL+"/media") { + t.Fatalf("直连模式应保留上游原始地址,实际 %q", track) + } + // 取媒体需要的防盗链头必须下发,否则客户端拿不到字节。 + if out.MediaHeaders["Referer"] == "" { + t.Fatalf("直连模式应下发 Referer,实际 %v", out.MediaHeaders) + } + if out.MediaHeaders["User-Agent"] == "" { + t.Fatalf("直连模式应下发 User-Agent,实际 %v", out.MediaHeaders) + } + // 传输层头交给客户端 HTTP 栈,不能下发。 + if _, ok := out.MediaHeaders["Accept-Encoding"]; ok { + t.Fatalf("不应下发 Accept-Encoding: %v", out.MediaHeaders) + } +} + +// TestMediaDirectAppliesToCachedChapter 缓存命中的章节同样要按本次请求决定通道。 +// +// 缓存里存的是书源侧原始地址(改写前),若命中后不改写就会把上游地址直接 +// 漏给代理模式的客户端,表现为「图片/音频打不开」。 +func TestMediaDirectAppliesToCachedChapter(t *testing.T) { + svc, _, book := prepareTicketBook(t) + ctx := t.Context() + + // 首次读取写入缓存(直连模式)。 + if err := svc.SetSourceMediaDirect(ctx, readerTestUserID, sourceIDOf(t, svc, book.Origin), true); err != nil { + t.Fatalf("开启直连失败: %v", err) + } + first, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{MediaDirect: true}) + if err != nil || !first.MediaDirect { + t.Fatalf("首次直连读取失败: %v", err) + } + + // 同章再按代理模式读:必须拿到签名代理地址,而不是缓存里的上游地址。 + second, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) + if err != nil { + t.Fatalf("二次读取失败: %v", err) + } + if second.MediaDirect { + t.Fatal("代理模式不应标记直连") + } + if !strings.HasPrefix(second.Tracks[0], "/api/reader/media?") { + t.Fatalf("缓存命中后仍应按代理改写,实际 %q", second.Tracks[0]) + } +} + +// TestMediaDirectForceProxyWins 客户端显式要求代理时,即使书源开着直连也走代理。 +// +// 这是直连取不到字节时的退路:不让用户为了继续读一章去改书源设置。 +func TestMediaDirectForceProxyWins(t *testing.T) { + svc, _, book := prepareTicketBook(t) + ctx := t.Context() + + if err := svc.SetSourceMediaDirect(ctx, readerTestUserID, sourceIDOf(t, svc, book.Origin), true); err != nil { + t.Fatalf("开启直连失败: %v", err) + } + + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, + ContentOptions{MediaDirect: true, ForceProxy: true}) + if err != nil { + t.Fatalf("读取失败: %v", err) + } + if out.MediaDirect { + t.Fatal("强制代理时不应标记直连") + } + if !strings.HasPrefix(out.Tracks[0], "/api/reader/media?") { + t.Fatalf("强制代理时应返回签名代理地址,实际 %q", out.Tracks[0]) + } + if len(out.MediaHeaders) != 0 { + t.Fatalf("代理模式不应下发请求头,实际 %v", out.MediaHeaders) + } +} + +// sourceIDOf 按书源 URL 取当前用户的书源 ID。 +func sourceIDOf(t *testing.T, svc *ReaderService, origin string) string { + t.Helper() + srcs, err := svc.ListSources(t.Context(), readerTestUserID) + if err != nil { + t.Fatalf("列书源失败: %v", err) + } + for _, src := range srcs { + if src.SourceURL == origin { + return src.ID + } + } + t.Fatalf("找不到书源 %s", origin) + return "" +} diff --git a/internal/service/reader/media_refresh_test.go b/internal/service/reader/media_refresh_test.go index fe32039..8a7de60 100644 --- a/internal/service/reader/media_refresh_test.go +++ b/internal/service/reader/media_refresh_test.go @@ -129,7 +129,7 @@ func TestRefreshMediaURLIssuesNewTicket(t *testing.T) { ctx := t.Context() // 首读:拿到第一张票并落缓存 - out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("首次读取失败: %v", err) } @@ -166,7 +166,7 @@ func TestRefreshMediaURLIssuesNewTicket(t *testing.T) { } // 缓存要被刷新成新票:再次读章节不应再回源 - out2, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + out2, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("换票后读章节失败: %v", err) } @@ -188,7 +188,7 @@ func TestRefreshMediaURLMemoryReusesTicket(t *testing.T) { svc, srv, book := prepareTicketBook(t) ctx := t.Context() - out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("首次读取失败: %v", err) } @@ -218,7 +218,7 @@ func TestRefreshMediaURLSkipsNonTrackURL(t *testing.T) { svc, _, book := prepareTicketBook(t) ctx := t.Context() - if _, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0); err != nil { + if _, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}); err != nil { t.Fatalf("首次读取失败: %v", err) } _, refreshable, err := svc.RefreshMediaURL(ctx, book, "https://img.example.com/cover.jpg") @@ -236,7 +236,7 @@ func TestRefreshMediaURLErrorsWhenSourceRepeatsTicket(t *testing.T) { svc, srv, book := prepareTicketBook(t) ctx := t.Context() - out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0) + out, err := svc.GetContentForBook(ctx, readerTestUserID, book.ID, 0, ContentOptions{}) if err != nil { t.Fatalf("首次读取失败: %v", err) } diff --git a/internal/service/reader/reader.go b/internal/service/reader/reader.go index eb7807f..2744147 100644 --- a/internal/service/reader/reader.go +++ b/internal/service/reader/reader.go @@ -1850,6 +1850,13 @@ type ChapterContent struct { // 前端据此决定走 hls.js 而不是原生