diff --git a/internal/handler/emby_playback.go b/internal/handler/emby_playback.go index 5102ce2..0450f7a 100644 --- a/internal/handler/emby_playback.go +++ b/internal/handler/emby_playback.go @@ -1,10 +1,13 @@ package handler import ( + "context" "errors" "net/http" "net/url" "strings" + "sync" + "time" "github.com/gin-gonic/gin" @@ -24,10 +27,98 @@ func embyPlaybackInfoHandler(svc *service.Container) gin.HandlerFunc { return } embyAttachRequestTokenToMediaSources(c, out) + // 在后台把本次条目的云盘直链换好:播放器拿到 PlaybackInfo 后通常还要 + // 1–2 秒才请求 /Videos/{id}/stream,把换链开销落在这段等待里。 + embyPrewarmPlaybackTargets(svc, c, out) c.JSON(http.StatusOK, out) } } +// embyPrewarmTimeout 是单次预热的等待上限。115 开放平台在跨太平洋线路上单次 +// 换链实测 0.4–1.1s,这里给足余量;超时只是没预热成功,不影响后续播放。 +const embyPrewarmTimeout = 10 * time.Second + +// embyPrewarmInFlight 去重同一个条目的并发预热(首页刷新会并发请求多个接口, +// 同一条目可能在短时间内被多次请求)。 +var embyPrewarmInFlight sync.Map + +// embyPrewarmSlots 限制同时进行的预热数量。客户端可能批量预取 PlaybackInfo +// (逐个剧集的预取请求),预热只是优化,不能反过来把 115 换链接口打出突发。 +// 名额满时直接跳过:排队等待的预热往往等真正播放时已经没意义了。 +var embyPrewarmSlots = make(chan struct{}, 4) + +// embyPrewarmPlaybackTargets 在后台预热本次 PlaybackInfo 涉及条目的云盘直链。 +// +// 起播链路里最贵的一步是「服务端拿 pickcode 去 115 开放平台换直链」:服务器在 +// 洛杉矶、115 接口在国内,冷启动实测 0.4–1.1s;之后 45 分钟内命中进程内缓存。 +// 播放器在 PlaybackInfo 与真正拉流之间有几秒间隔,这里把换链放到那段间隔里, +// 起播时就只剩纯网络耗时。 +// +// 只处理云盘/strm 条目,且失败一律静默忽略:预热是尽力而为的优化,不能影响 +// PlaybackInfo 的正常返回。 +func embyPrewarmPlaybackTargets(svc *service.Container, c *gin.Context, out map[string]any) { + if svc == nil || svc.Strm == nil || svc.Repo == nil || svc.Repo.Media == nil || out == nil { + return + } + ids := embyPrewarmMediaIDs(out) + if len(ids) == 0 { + return + } + userAgent := c.GetHeader("User-Agent") + // 预热是给「后续请求」用的:即便本次 PlaybackInfo 的连接断开, + // 也要把换链跑完。 + base := context.WithoutCancel(c.Request.Context()) + for _, mediaID := range ids { + if _, loaded := embyPrewarmInFlight.LoadOrStore(mediaID, struct{}{}); loaded { + continue + } + go func(id string) { + defer embyPrewarmInFlight.Delete(id) + select { + case embyPrewarmSlots <- struct{}{}: + defer func() { <-embyPrewarmSlots }() + default: + return + } + ctx, cancel := context.WithTimeout(base, embyPrewarmTimeout) + defer cancel() + m, err := svc.Repo.Media.FindByID(ctx, id) + if err != nil || m == nil { + return + } + raw := strings.TrimSpace(m.STRMURL) + if raw == "" || !service.IsStrmMediaRow(m) { + return + } + // 解析结果由 strm 层按 pickcode+UA 缓存;已缓存时这里是空转。 + _, _ = svc.Strm.ResolvePlayTargetWithUA(ctx, raw, userAgent) + }(mediaID) + } +} + +// embyPrewarmMediaIDs 取出 PlaybackInfo 载荷里 MediaSources 的条目 ID。 +func embyPrewarmMediaIDs(out map[string]any) []string { + sources, ok := out["MediaSources"].([]map[string]any) + if !ok || len(sources) == 0 { + return nil + } + ids := make([]string, 0, len(sources)) + seen := make(map[string]struct{}, len(sources)) + for _, src := range sources { + id, _ := src["Id"].(string) + id = strings.TrimSpace(id) + if id == "" { + continue + } + if _, dup := seen[id]; dup { + continue + } + seen[id] = struct{}{} + ids = append(ids, id) + } + return ids +} + // embySubtitleStreamHandler serves an external subtitle track advertised in a // MediaSource's MediaStreams via its Emby index // (/Videos/:id/Subtitles/:index/Stream). The index maps to a discovered diff --git a/internal/handler/emby_playback_prewarm_test.go b/internal/handler/emby_playback_prewarm_test.go new file mode 100644 index 0000000..80bdcce --- /dev/null +++ b/internal/handler/emby_playback_prewarm_test.go @@ -0,0 +1,159 @@ +package handler + +import ( + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/glebarez/sqlite" + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/repository" + "github.com/truewhile/MeBox/internal/service" +) + +func newPrewarmTestContainer(t *testing.T) *service.Container { + t.Helper() + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatalf("open db: %v", err) + } + if err := db.AutoMigrate(&model.Media{}, &model.Setting{}, &model.StrmAccount{}); err != nil { + t.Fatalf("migrate: %v", err) + } + if sqlDB, err := db.DB(); err == nil { + // 内存库 + 后台预热协程:限制单连接,避免新连接拿到空白的 :memory:。 + sqlDB.SetMaxOpenConns(1) + } + repos := repository.New(db) + return &service.Container{ + Log: zap.NewNop(), + Repo: repos, + Strm: service.NewStrmService(&config.Config{}, zap.NewNop(), repos, nil), + } +} + +func newPrewarmTestContext() *gin.Context { + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodPost, "/emby/Items/media-1/PlaybackInfo", nil) + c.Request.Header.Set("User-Agent", "RodelPlayer/2.2607.7.0") + return c +} + +func TestEmbyPrewarmMediaIDsExtractsDeduplicates(t *testing.T) { + out := map[string]any{ + "MediaSources": []map[string]any{ + {"Id": "src-1"}, + {"Id": " src-1 "}, + {"Id": "src-2"}, + {"Id": ""}, + {"Name": "no id"}, + }, + } + got := embyPrewarmMediaIDs(out) + if len(got) != 2 || got[0] != "src-1" || got[1] != "src-2" { + t.Fatalf("ids = %v, want [src-1 src-2]", got) + } + if got := embyPrewarmMediaIDs(map[string]any{}); len(got) != 0 { + t.Fatalf("missing MediaSources should yield no ids, got %v", got) + } + if got := embyPrewarmMediaIDs(map[string]any{"MediaSources": []any{}}); len(got) != 0 { + t.Fatalf("foreign payload shape should yield no ids, got %v", got) + } +} + +// 预热是异步的:调用必须立即返回,并且协程结束后不能残留去重标记。 +func TestEmbyPrewarmPlaybackTargetsRunsAsyncAndCleansUp(t *testing.T) { + svc := newPrewarmTestContainer(t) + if err := svc.Repo.DB.Create(&model.Media{ + Base: model.Base{ID: "media-1"}, + Title: "Cloud", + Path: "cloud://cloud115/Movie.mkv", + Container: "strm", + STRMURL: "/api/strm/play/cloud115/video.mkv?acct=missing&pickcode=pc1", + }).Error; err != nil { + t.Fatal(err) + } + + embyPrewarmInFlight.Delete("media-1") + out := map[string]any{"MediaSources": []map[string]any{{"Id": "media-1"}}} + + done := make(chan struct{}) + go func() { + embyPrewarmPlaybackTargets(svc, newPrewarmTestContext(), out) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("embyPrewarmPlaybackTargets blocked the caller") + } + + // 后台协程应很快跑完并释放去重标记,否则同一条目后续再也预热不了。 + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if _, busy := embyPrewarmInFlight.Load("media-1"); !busy { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("prewarm in-flight marker leaked") +} + +// 各种缺数据的情况都不允许 panic 或阻塞:预热只是尽力而为的优化。 +func TestEmbyPrewarmPlaybackTargetsIsNilSafe(t *testing.T) { + svc := newPrewarmTestContainer(t) + c := newPrewarmTestContext() + out := map[string]any{"MediaSources": []map[string]any{{"Id": "media-1"}}} + + cases := []struct { + name string + svc *service.Container + out map[string]any + }{ + {name: "空容器", svc: &service.Container{}, out: out}, + {name: "无 Strm", svc: &service.Container{Repo: svc.Repo}, out: out}, + {name: "无 Repo", svc: &service.Container{Strm: svc.Strm}, out: out}, + {name: "nil 载荷", svc: svc, out: nil}, + {name: "无 MediaSources", svc: svc, out: map[string]any{}}, + {name: "条目不存在", svc: svc, out: out}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + embyPrewarmPlaybackTargets(tc.svc, c, tc.out) + }) + } +} + +// 本地文件条目不该触发换链预热(没有云盘直链可预热)。 +func TestEmbyPrewarmSkipsLocalMedia(t *testing.T) { + svc := newPrewarmTestContainer(t) + if err := svc.Repo.DB.Create(&model.Media{ + Base: model.Base{ID: "local-1"}, + Title: "Local", + Path: "/media/movies/Local.mkv", + LibraryID: "lib-1", + }).Error; err != nil { + t.Fatal(err) + } + + embyPrewarmInFlight.Delete("local-1") + embyPrewarmPlaybackTargets(svc, newPrewarmTestContext(), + map[string]any{"MediaSources": []map[string]any{{"Id": "local-1"}}}) + + // 协程要么已经跑完(标记被清掉),要么根本没起;两种都不该留下标记。 + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if _, busy := embyPrewarmInFlight.Load("local-1"); !busy { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("local media must not leave a prewarm marker") +} diff --git a/internal/service/emby_discovery.go b/internal/service/emby_discovery.go index 8e7e43e..296e232 100644 --- a/internal/service/emby_discovery.go +++ b/internal/service/emby_discovery.go @@ -5,6 +5,7 @@ import ( "crypto/sha1" "encoding/hex" "strings" + "time" ) // Emby 发现类接口:NextUp / Similar / Genres。 @@ -67,6 +68,19 @@ func (e *EmbyService) SimilarItems(ctx context.Context, mediaID, userID string, return emptyItemsEnvelope(0), nil } + // 详情页每次打开都会请求相似推荐,而重建要走「取候选池 + 内存打分」 + // (实测冷 340ms / 热 70ms)。推荐列表短暂陈旧无害,用短 TTL 缓存, + // 新建库或换用户都会因为键名不同而自然隔离。 + cacheKey := e.embySimilarCacheKey(mediaID, userID, limit) + if e.cache != nil { + var cached map[string]any + if e.cache.GetJSON(ctx, cacheKey, &cached) && cached != nil { + if _, ok := cached["Items"]; ok { + return cached, nil + } + } + } + // Bug 2 fix: resolve virtual series IDs (msgo-series-*) and real series // table IDs to a representative episode so SimilarCandidates (which calls // Media.FindByID) can seed similarity from concrete media metadata. @@ -109,12 +123,20 @@ func (e *EmbyService) SimilarItems(ctx context.Context, mediaID, userID string, if err != nil { return nil, err } - return map[string]any{ + out := map[string]any{ "Items": items, "TotalRecordCount": int64(len(items)), - }, nil + } + if e.cache != nil { + e.cache.SetJSON(ctx, cacheKey, out, embySimilarCacheTTL) + } + return out, nil } +// embySimilarCacheTTL 是「相似推荐」结果的缓存时长。列表只是推荐,短暂陈旧 +// 无害;TTL 取短一些,让新入库的内容尽快出现。 +const embySimilarCacheTTL = 2 * time.Minute + // Genres 返回类型清单。parentID 非空时(客户端按媒体库浏览类型)只统计该库。 func (e *EmbyService) Genres(ctx context.Context, userID, parentID string) (map[string]any, error) { discovery := e.discoveryService() diff --git a/internal/service/emby_items_cache.go b/internal/service/emby_items_cache.go index a31a030..83b5716 100644 --- a/internal/service/emby_items_cache.go +++ b/internal/service/emby_items_cache.go @@ -67,6 +67,15 @@ func (e *EmbyService) embyLatestCacheKey(userID, parentID string, limit int) str return "media:emby:" + hex.EncodeToString(sum[:]) } +// embySimilarCacheKey 是「相似推荐」结果的缓存键。 +// +// userID 必须参与键名:候选集的可见性(AllowedLibraryIDs、NSFW)由用户决定, +// 混用会把别的用户可见的条目推荐给当前用户。limit 同理影响结果条数与排序。 +func (e *EmbyService) embySimilarCacheKey(mediaID, userID string, limit int) string { + sum := sha256.Sum256([]byte(strings.Join([]string{"similar-v1", mediaID, userID, strconv.Itoa(limit)}, "|"))) + return "media:emby:" + hex.EncodeToString(sum[:]) +} + // defaultEmbyLatestCacheTTLSeconds 是 Emby「最新添加」缓存的兜底时长。 const defaultEmbyLatestCacheTTLSeconds = 300 diff --git a/internal/service/emby_remote.go b/internal/service/emby_remote.go index 0916bbd..19ec3b2 100644 --- a/internal/service/emby_remote.go +++ b/internal/service/emby_remote.go @@ -86,6 +86,54 @@ type EmbyRemoteService struct { // (没有它时 URL 恒定,缩略图会永久停留在旧版本)。 imageTagMu sync.RWMutex imageTags map[string]string + + // remoteGate 是发往远程 Emby 的并发闸门。第三方客户端刷新首页时会为每个 + // 远程媒体库各请求一次 /Items/Latest,挂着几十个库就是几十路并发(生产环境 + // 实测 50 路同时打进来,单个请求被拖到 5s+)。限制在途请求数后单个请求的 + // 等待时间反而下降,也不会把 2C 小机和对方服务器一起打满。 + // + // nil 表示不限流(测试直接构造结构体时走这条路)。 + remoteGate chan struct{} +} + +// embyRemoteConcurrencyLimit 是同时发往远程 Emby 的请求数上限。 +const embyRemoteConcurrencyLimit = 8 + +// enterRemoteGate 取得一个远程请求名额,返回释放函数。未配置闸门时返回空操作。 +func (r *EmbyRemoteService) enterRemoteGate(ctx context.Context) (func(), error) { + if r == nil || r.remoteGate == nil { + return func() {}, nil + } + select { + case r.remoteGate <- struct{}{}: + return func() { <-r.remoteGate }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// fetchRemoteBody 在并发闸门内发起请求并读完响应体,返回状态码与字节。 +func (r *EmbyRemoteService) fetchRemoteBody(ctx context.Context, req *http.Request, path string) (int, []byte, error) { + release, err := r.enterRemoteGate(ctx) + if err != nil { + return 0, nil, err + } + defer release() + resp, err := r.http.Do(req) + if err != nil { + return 0, nil, redactSensitiveError(fmt.Errorf("请求远程 Emby 失败: %w", err)) + } + defer resp.Body.Close() + // 读 8MB+1 以区分"刚好 8MB"与"被截断":截断的 JSON 会让 + // Unmarshal 报 unexpected end,难以定位;这里显式报错。 + data, readErr := io.ReadAll(io.LimitReader(resp.Body, (8<<20)+1)) + if readErr != nil { + return resp.StatusCode, nil, readErr + } + if len(data) > 8<<20 { + return resp.StatusCode, data, fmt.Errorf("远程 Emby 响应超过 8MB 上限(路径 %s):请减小分页或 Fields 字段", path) + } + return resp.StatusCode, data, nil } type embyRemotePersonImageRef struct { @@ -113,6 +161,7 @@ func NewEmbyRemoteService(cfg *config.Config, log *zap.Logger, repo *repository. stream: &http.Client{ Transport: &embyRemoteTransport{base: http.DefaultTransport}, }, + remoteGate: make(chan struct{}, embyRemoteConcurrencyLimit), } } @@ -645,21 +694,11 @@ func (r *EmbyRemoteService) doGetOnLine(ctx context.Context, acct *model.StrmAcc return err } req.Header.Set("X-Emby-Token", cfg.Token) - resp, err := r.http.Do(req) + status, data, err := r.fetchRemoteBody(ctx, req, path) if err != nil { - return redactSensitiveError(fmt.Errorf("请求远程 Emby 失败: %w", err)) + return err } - // 读 8MB+1 以区分"刚好 8MB"与"被截断":截断的 JSON 会让 - // Unmarshal 报 unexpected end,难以定位;这里显式报错。 - data, readErr := io.ReadAll(io.LimitReader(resp.Body, (8<<20)+1)) - resp.Body.Close() - if readErr != nil { - return readErr - } - if len(data) > 8<<20 { - return fmt.Errorf("远程 Emby 响应超过 8MB 上限(路径 %s):请减小分页或 Fields 字段", path) - } - if resp.StatusCode == http.StatusUnauthorized && attempt == 0 { + if status == http.StatusUnauthorized && attempt == 0 { // 401:只清当前线路的内存 token 并立即重认证;不在此时删除 // DB 里的 api_key——①外层还会按线路故障转移(其他线路可能 // 存有自己的 token);②纯 api_key 账号删除后无法再认证,一次 @@ -673,8 +712,8 @@ func (r *EmbyRemoteService) doGetOnLine(ctx context.Context, acct *model.StrmAcc master.RemoteUserID = cfg.RemoteUserID continue } - if resp.StatusCode >= 300 { - return redactSensitiveError(fmt.Errorf("远程 Emby 请求失败(%d): %s", resp.StatusCode, strings.TrimSpace(string(data)))) + if status >= 300 { + return redactSensitiveError(fmt.Errorf("远程 Emby 请求失败(%d): %s", status, strings.TrimSpace(string(data)))) } if out == nil { return nil @@ -1617,6 +1656,12 @@ func (r *EmbyRemoteService) doMutateOnLine(ctx context.Context, cfg *EmbyRemoteC return err } req.Header.Set("X-Emby-Token", cfg.Token) + // 状态同步同样走远程并发闸门:它和首页那批 Latest 请求共用对方服务器。 + release, err := r.enterRemoteGate(ctx) + if err != nil { + return err + } + defer release() resp, err := r.http.Do(req) if err != nil { return redactSensitiveError(fmt.Errorf("请求远程 Emby 失败: %w", err)) diff --git a/internal/service/emby_remote_gate_test.go b/internal/service/emby_remote_gate_test.go new file mode 100644 index 0000000..50e213a --- /dev/null +++ b/internal/service/emby_remote_gate_test.go @@ -0,0 +1,112 @@ +package service + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" +) + +// 远程 Emby 的并发闸门必须真的把在途请求数压在上限之内:第三方客户端首页会为 +// 每个远程媒体库各请求一次 /Items/Latest,几十个库就是几十路并发。 +func TestRemoteGateLimitsConcurrentRequests(t *testing.T) { + const gate = 3 + const requests = 12 + + var mu sync.Mutex + inflight, peak := 0, 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + mu.Lock() + inflight++ + if inflight > peak { + peak = inflight + } + mu.Unlock() + + time.Sleep(20 * time.Millisecond) + + mu.Lock() + inflight-- + mu.Unlock() + _, _ = w.Write([]byte(`{"Items":[]}`)) + })) + defer srv.Close() + + svc := &EmbyRemoteService{http: srv.Client(), remoteGate: make(chan struct{}, gate)} + + var wg sync.WaitGroup + for i := 0; i < requests; i++ { + wg.Add(1) + go func() { + defer wg.Done() + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL+"/Items/Latest", nil) + if err != nil { + return + } + if _, _, err := svc.fetchRemoteBody(context.Background(), req, "/Items/Latest"); err != nil { + t.Errorf("fetchRemoteBody: %v", err) + } + }() + } + wg.Wait() + + mu.Lock() + defer mu.Unlock() + if peak == 0 { + t.Fatal("test server never saw a request") + } + if peak > gate { + t.Fatalf("peak concurrency = %d, want <= %d", peak, gate) + } +} + +// 闸门排队时要响应请求取消,不能把整个 HTTP 请求挂死。 +func TestRemoteGateHonoursContextCancellation(t *testing.T) { + svc := &EmbyRemoteService{remoteGate: make(chan struct{}, 1)} + svc.remoteGate <- struct{}{} // 占满名额 + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := svc.enterRemoteGate(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + + // 释放名额后必须能正常拿到。 + <-svc.remoteGate + release, err := svc.enterRemoteGate(context.Background()) + if err != nil { + t.Fatalf("enterRemoteGate after release: %v", err) + } + release() + if len(svc.remoteGate) != 0 { + t.Fatalf("gate leaked a permit: len = %d", len(svc.remoteGate)) + } +} + +// 未配置闸门(测试里直接构造结构体)时不应限流,保持旧行为。 +func TestRemoteGateAbsentIsUnlimited(t *testing.T) { + svc := &EmbyRemoteService{} + release, err := svc.enterRemoteGate(context.Background()) + if err != nil { + t.Fatalf("enterRemoteGate: %v", err) + } + release() +} + +// 生产路径构造出来的服务必须带闸门,否则上面的限制形同虚设。 +func TestNewEmbyRemoteServiceInitialisesGate(t *testing.T) { + svc := NewEmbyRemoteService(&config.Config{}, zap.NewNop(), nil, nil) + if svc.remoteGate == nil { + t.Fatal("remoteGate must be initialised by the constructor") + } + if cap(svc.remoteGate) != embyRemoteConcurrencyLimit { + t.Fatalf("gate capacity = %d, want %d", cap(svc.remoteGate), embyRemoteConcurrencyLimit) + } +} diff --git a/internal/service/emby_similar_cache_test.go b/internal/service/emby_similar_cache_test.go new file mode 100644 index 0000000..c3f6961 --- /dev/null +++ b/internal/service/emby_similar_cache_test.go @@ -0,0 +1,118 @@ +package service + +import ( + "context" + "testing" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/model" +) + +// totalRecordCount 兼容两种来源:直算出来的是 int64,经过进程内缓存 JSON +// 往返后是 float64。两者序列化成 Emby 响应时完全一致。 +func totalRecordCount(t *testing.T, out map[string]any) int { + t.Helper() + switch v := out["TotalRecordCount"].(type) { + case int64: + return int(v) + case int: + return v + case float64: + return int(v) + default: + t.Fatalf("unexpected TotalRecordCount type %T (%v)", out["TotalRecordCount"], out["TotalRecordCount"]) + return 0 + } +} + +// 相似推荐的结果要缓存:详情页每次打开都会请求它,重建要走「取候选池 + 内存打分」 +// (实测冷 340ms / 热 70ms)。 +func TestSimilarItemsServesCachedPayload(t *testing.T) { + svc := newTestEmbyService(t) + repos := svc.repo + svc.SetRuntimeCache(NewRuntimeCacheService(&config.Config{}, zap.NewNop())) + + libID := seedDiscoveryLibrary(t, repos, "movie") + source := seedSimilarMedia(t, repos, libID, "源片", "Action", 2010, 7) + seedSimilarMedia(t, repos, libID, "候选甲", "Action", 2010, 7) + seedSimilarMedia(t, repos, libID, "候选乙", "Comedy", 2011, 6) + + ctx := context.Background() + first, err := svc.SimilarItems(ctx, source.ID, "user-1", 10) + if err != nil { + t.Fatalf("first SimilarItems: %v", err) + } + want := totalRecordCount(t, first) + if want == 0 { + t.Fatalf("first call returned no candidates, test data is wrong: %+v", first) + } + + // 把候选全部删掉:第二次如果还返回原结果,只可能是命中缓存。 + if err := repos.DB.Where("id <> ?", source.ID).Delete(&model.Media{}).Error; err != nil { + t.Fatal(err) + } + + second, err := svc.SimilarItems(ctx, source.ID, "user-1", 10) + if err != nil { + t.Fatalf("second SimilarItems: %v", err) + } + if got := totalRecordCount(t, second); got != want { + t.Fatalf("cached call returned %d items, want %d (cache miss?)", got, want) + } + + // 另一个用户(不同的可见性)不能复用别人的缓存:这里应当重新查询并得到 0。 + other, err := svc.SimilarItems(ctx, source.ID, "user-2", 10) + if err != nil { + t.Fatalf("other user SimilarItems: %v", err) + } + if got := totalRecordCount(t, other); got != 0 { + t.Fatalf("other user got %d items, want 0 (per-user cache key)", got) + } +} + +// limit 不同必须分开缓存,否则一次小 limit 请求会污染后续更大的请求。 +func TestSimilarItemsCacheSeparatesLimit(t *testing.T) { + svc := newTestEmbyService(t) + repos := svc.repo + svc.SetRuntimeCache(NewRuntimeCacheService(&config.Config{}, zap.NewNop())) + + libID := seedDiscoveryLibrary(t, repos, "movie") + source := seedSimilarMedia(t, repos, libID, "源片", "Action", 2010, 7) + for _, title := range []string{"甲", "乙", "丙", "丁"} { + seedSimilarMedia(t, repos, libID, "候选"+title, "Action", 2010, 7) + } + + ctx := context.Background() + small, err := svc.SimilarItems(ctx, source.ID, "user-1", 2) + if err != nil { + t.Fatal(err) + } + if got := totalRecordCount(t, small); got != 2 { + t.Fatalf("limit=2 returned %d items, want 2", got) + } + large, err := svc.SimilarItems(ctx, source.ID, "user-1", 4) + if err != nil { + t.Fatal(err) + } + if got := totalRecordCount(t, large); got != 4 { + t.Fatalf("limit=4 returned %d items, want 4 (limit must be part of the cache key)", got) + } +} + +// 没有注入缓存时(测试/精简部署)也必须正常工作。 +func TestSimilarItemsWithoutCacheStillWorks(t *testing.T) { + svc := newTestEmbyService(t) + libID := seedDiscoveryLibrary(t, svc.repo, "movie") + source := seedSimilarMedia(t, svc.repo, libID, "源片", "Action", 2010, 7) + seedSimilarMedia(t, svc.repo, libID, "候选甲", "Action", 2010, 7) + + out, err := svc.SimilarItems(context.Background(), source.ID, "user-1", 10) + if err != nil { + t.Fatalf("SimilarItems: %v", err) + } + if got := totalRecordCount(t, out); got == 0 { + t.Fatalf("expected candidates without a cache, got %+v", out) + } +} diff --git a/internal/service/service_builder.go b/internal/service/service_builder.go index 901fa3e..ef2d96c 100644 --- a/internal/service/service_builder.go +++ b/internal/service/service_builder.go @@ -145,6 +145,9 @@ func (b *serviceContainerBuilder) initContentServices() { b.c.Transcoder.SetStrmPlayTargetResolver(b.c.Strm.ResolvePlayTarget) b.c.Transcoder.SetProbe(b.c.FFprobe) b.c.Subtitle.SetStrmPlayTargetResolver(b.c.Strm.ResolvePlayTarget) + // 播放链路:/Videos/{id}/stream 与 /api/stream/{id} 在服务端完成换链后直接 + // 302 到最终直链,客户端少跟随一次 302(高延迟线路上省一个往返)。 + b.c.Stream.SetStrmPlayTargetResolver(b.c.Strm.ResolvePlayTargetWithUA) // 弹幕识别需要把远程 Emby 条目解析为 Media 元数据及可拉取前 16MB 的直链 URL。 if b.c.EmbyRemote != nil { b.c.Danmaku.SetRemoteMediaResolver(func(ctx context.Context, encodedID string) (*model.Media, string, error) { diff --git a/internal/service/stream_direct_redirect_test.go b/internal/service/stream_direct_redirect_test.go new file mode 100644 index 0000000..c1fefd2 --- /dev/null +++ b/internal/service/stream_direct_redirect_test.go @@ -0,0 +1,200 @@ +package service + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/repository" + "github.com/truewhile/MeBox/internal/service/cloud" +) + +// directRedirectTestRepo 建一个带网盘账号表的库:normalizeCloudPlayTarget 需要 +// StrmAccount 才能判断 strm 目标是不是本机账号生成的。 +func directRedirectTestRepo(t *testing.T) *repository.Container { + t.Helper() + return repository.New(newServiceTestDB(t, &model.Media{}, &model.Setting{}, &model.StrmAccount{})) +} + +func seedCloudSTRMMedia(t *testing.T, repos *repository.Container, id, strmURL string) { + t.Helper() + if err := repos.DB.Create(&model.Media{ + Base: model.Base{ID: id}, + Title: "Cloud", + Path: "cloud://cloud115/Movie.mkv", + Container: "strm", + STRMURL: strmURL, + }).Error; err != nil { + t.Fatal(err) + } +} + +// 服务端能换到最终直链时必须直接 302 过去:客户端原本要跟着 +// /Videos/{id}/stream → /api/strm/play 两次 302,现在缩成一跳。 +func TestServeFileRedirectsStraightToResolvedDirectURL(t *testing.T) { + repos := directRedirectTestRepo(t) + seedCloudSTRMMedia(t, repos, "cloud-direct", "/api/strm/play/cloud115/video.mkv?acct=a1&pickcode=pc1") + direct := "https://cdnfhnfile.115cdn.net/637b/Movie.mkv?t=1&k=sig" + + var gotRaw, gotUA string + svc := NewStreamService(&config.Config{}, zap.NewNop(), repos, nil). + SetStrmPlayTargetResolver(func(_ context.Context, raw, userAgent string) (*StrmPlayResult, error) { + gotRaw, gotUA = raw, userAgent + // 生产环境 115 直链就是这样返回的:绑定 UA、Proxy=false。 + return &StrmPlayResult{ + RedirectURL: direct, + Link: &cloud.DirectLink{URL: direct, Headers: map[string]string{"User-Agent": userAgent}}, + }, nil + }) + + req := httptest.NewRequest(http.MethodGet, "http://nas.local:18080/api/stream/cloud-direct?token=jwt123", nil) + req.Header.Set("User-Agent", "RodelPlayer/2.2607.7.0") + w := httptest.NewRecorder() + + if err := svc.ServeFile(w, req, "cloud-direct"); err != nil { + t.Fatalf("ServeFile: %v", err) + } + if w.Code != http.StatusFound { + t.Fatalf("status = %d, want 302", w.Code) + } + loc := w.Header().Get("Location") + if loc != direct { + t.Fatalf("Location = %q, want the resolved direct link %q", loc, direct) + } + if strings.Contains(loc, "jwt123") || strings.Contains(loc, "media_id=") { + t.Fatalf("internal auth query must not leak to the CDN link: %q", loc) + } + if got := w.Header().Get("Cache-Control"); !strings.Contains(got, "no-store") { + t.Fatalf("direct redirect must stay uncacheable, got %q", got) + } + // 换链必须带播放器 UA(115 直链绑定换取时的 UA,且按 UA 分键缓存)。 + if gotUA != "RodelPlayer/2.2607.7.0" { + t.Fatalf("resolver UA = %q, want the player UA", gotUA) + } + if !strings.Contains(gotRaw, "pickcode=pc1") { + t.Fatalf("resolver raw = %q, want the strm target", gotRaw) + } +} + +// 只要拿不到「客户端自己能直接拉取的直链」,就必须回退到改动前的 strm 端点跳转, +// 保证行为不会比改动前更差。 +func TestServeFileFallsBackWhenDirectResolveUnavailable(t *testing.T) { + strmURL := "/api/strm/play/cloud115/video.mkv?acct=a1&pickcode=pc1" + cases := []struct { + name string + result *StrmPlayResult + wantErr error + }{ + {name: "换链失败", wantErr: errors.New("115 换链失败")}, + { + name: "需要服务端反向代理", + result: &StrmPlayResult{ + Proxy: true, + Link: &cloud.DirectLink{URL: "https://cdn.example/x", Headers: map[string]string{"Authorization": "Basic x"}}, + }, + }, + {name: "没有直链(别的 MeBox 实例)", result: &StrmPlayResult{RedirectURL: "https://other.example/api/strm/play/cloud115/video.mkv?acct=o&pickcode=p"}}, + {name: "解析到本地文件", result: &StrmPlayResult{LocalPath: "/media/Movie.mkv"}}, + {name: "返回 nil", result: nil}, + { + name: "链接要求额外请求头", + result: &StrmPlayResult{ + RedirectURL: "https://cdn.example/x", + Link: &cloud.DirectLink{ + URL: "https://cdn.example/x", + Headers: map[string]string{"User-Agent": "ua", "Authorization": "Bearer t"}, + }, + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + repos := directRedirectTestRepo(t) + seedCloudSTRMMedia(t, repos, "cloud-fallback", strmURL) + svc := NewStreamService(&config.Config{}, zap.NewNop(), repos, nil). + SetStrmPlayTargetResolver(func(context.Context, string, string) (*StrmPlayResult, error) { + return tc.result, tc.wantErr + }) + + req := httptest.NewRequest(http.MethodGet, "http://nas.local:18080/api/stream/cloud-fallback?token=jwt123", nil) + w := httptest.NewRecorder() + + if err := svc.ServeFile(w, req, "cloud-fallback"); err != nil { + t.Fatalf("ServeFile: %v", err) + } + if w.Code != http.StatusFound { + t.Fatalf("status = %d, want 302", w.Code) + } + loc := w.Header().Get("Location") + if !strings.Contains(loc, "/api/strm/play/cloud115/video.mkv") { + t.Fatalf("Location = %q, want the strm endpoint fallback", loc) + } + if strings.Contains(loc, "cdn.example") || strings.Contains(loc, "other.example") { + t.Fatalf("Location = %q, must not point at an unusable direct link", loc) + } + }) + } +} + +// 没有注入解析器(测试/精简部署)时保持原有跳转,不受本次优化影响。 +func TestServeFileKeepsSTRMEndpointWithoutResolver(t *testing.T) { + repos := directRedirectTestRepo(t) + seedCloudSTRMMedia(t, repos, "cloud-plain", "/api/strm/play/cloud115/video.mkv?acct=a1&pickcode=pc1") + svc := NewStreamService(&config.Config{}, zap.NewNop(), repos, nil) + + req := httptest.NewRequest(http.MethodGet, "http://nas.local:18080/api/stream/cloud-plain?token=jwt123", nil) + w := httptest.NewRecorder() + if err := svc.ServeFile(w, req, "cloud-plain"); err != nil { + t.Fatalf("ServeFile: %v", err) + } + if loc := w.Header().Get("Location"); !strings.Contains(loc, "/api/strm/play/cloud115/video.mkv") { + t.Fatalf("Location = %q, want the strm endpoint", loc) + } +} + +// playbackQueryWithUA 负责把播放器 UA 透传给换链方,同时不能改动原 URL。 +func TestPlaybackQueryWithUAInjectsUserAgent(t *testing.T) { + u, err := url.Parse("/api/strm/play/cloud115/video.mkv?acct=a1&pickcode=pc1") + if err != nil { + t.Fatal(err) + } + q := playbackQueryWithUA(u, " RodelPlayer/2.2607.7.0 ") + if got := q.Get("__ua"); got != "RodelPlayer/2.2607.7.0" { + t.Fatalf("__ua = %q, want the trimmed player UA", got) + } + if q.Get("pickcode") != "pc1" || q.Get("acct") != "a1" { + t.Fatalf("original query lost: %v", q) + } + if strings.Contains(u.RawQuery, "__ua") { + t.Fatalf("source URL must not be mutated: %q", u.RawQuery) + } + if got := playbackQueryWithUA(u, " ").Get("__ua"); got != "" { + t.Fatalf("blank UA must not be injected, got %q", got) + } + if got := playbackQueryWithUA(nil, "ua").Get("__ua"); got != "ua" { + t.Fatalf("nil URL must still accept the UA, got %q", got) + } +} + +// ResolvePlayTargetWithUA 是 ResolvePlayTarget 的 UA 版本:空 UA 时行为必须与 +// 原方法完全一致(外部直链透传)。 +func TestResolvePlayTargetWithUAKeepsPassthroughBehaviour(t *testing.T) { + svc := &StrmService{} + raw := "https://cdn.example.test/Movie.mkv?sign=1" + got, err := svc.ResolvePlayTargetWithUA(context.Background(), raw, "RodelPlayer/1.0") + if err != nil { + t.Fatalf("ResolvePlayTargetWithUA: %v", err) + } + if got == nil || got.RedirectURL != raw { + t.Fatalf("result = %+v, want passthrough of %q", got, raw) + } +} diff --git a/internal/service/stream_file.go b/internal/service/stream_file.go index ba53cde..570c904 100644 --- a/internal/service/stream_file.go +++ b/internal/service/stream_file.go @@ -6,6 +6,7 @@ import ( "net/url" "os" "strings" + "time" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" @@ -33,6 +34,14 @@ func (s *StreamService) ServeFileWithCloudMode(w http.ResponseWriter, r *http.Re if !cloudPlaybackModeEnabled(r.Context(), s.repo, cloudMode) { return ErrCloudPlaybackDisabled } + // 能在服务端换到最终直链就直接 302 过去:客户端少跟随一次 302,等于 + // 省掉一次「DNS+TCP+TLS+请求」的往返。解析结果同时写进 strm 层直链 + // 缓存,后续 /api/strm/play 请求直接命中。 + if direct, ok := s.resolveDirectPlayTargetURL(r, strmURL); ok { + setCloudRedirectNoStore(w) + http.Redirect(w, r, direct, http.StatusFound) + return nil + } // 云盘播放 URL 先规范化为相对路径,免疫扫描时固化的旧 host; // 指向别的 MeBox 实例的地址保持原样,按第三方直链透传。 target := normalizeCloudPlayTarget(r.Context(), s.repo, s.cfg, r, strmURL) @@ -72,6 +81,43 @@ func setCloudRedirectNoStore(w http.ResponseWriter) { w.Header().Set("Expires", "0") } +// directPlayResolveTimeout 是服务端换链的等待上限。115 开放平台在跨太平洋线路 +// 上单次换链实测 0.4–1.1s,这里给足余量;一旦超时就回退到原来的 strm 端点跳转, +// 由 /api/strm/play 再去换链,最坏情况只是回到改动前的行为。 +const directPlayResolveTimeout = 10 * time.Second + +// resolveDirectPlayTargetURL 尝试在服务端把 strm 目标解析成客户端可直接拉取的 +// 最终直链,供调用方直接 302。 +// +// 只在「明确的直链」上短路:需要服务端反向代理(云盘 WebDAV 等必须附加请求头)、 +// 解析到本地文件、以及解析失败都会返回 false,由调用方按改动前的方式回退到 +// strm 端点跳转,行为不会变差。 +func (s *StreamService) resolveDirectPlayTargetURL(r *http.Request, raw string) (string, bool) { + if s == nil || s.strmResolve == nil || r == nil { + return "", false + } + ctx, cancel := context.WithTimeout(r.Context(), directPlayResolveTimeout) + defer cancel() + // 必须带播放器 UA:115 直链绑定换取时的 UA,且按 UA 分键缓存,换错会拿到 + // 与播放器不匹配(或未命中缓存)的地址。 + result, err := s.strmResolve(ctx, raw, r.Header.Get("User-Agent")) + if err != nil || result == nil || result.Proxy || result.Link == nil { + return "", false + } + // 客户端只能自带 User-Agent 这类基础请求头。链接一旦要求其它头(Referer / + // Authorization),就必须继续由服务端反向代理,不能在这里短路。 + for name := range result.Link.Headers { + if !strings.EqualFold(strings.TrimSpace(name), "User-Agent") { + return "", false + } + } + direct := strings.TrimSpace(result.RedirectURL) + if direct == "" { + return "", false + } + return direct, true +} + func isCloudPlaybackTarget(raw string) bool { _, _, ok := parseCloudMediaPlaybackURL(raw) return ok diff --git a/internal/service/stream_service.go b/internal/service/stream_service.go index e44d6ea..1a927c9 100644 --- a/internal/service/stream_service.go +++ b/internal/service/stream_service.go @@ -33,6 +33,10 @@ type StreamService struct { log *zap.Logger repo *repository.Container transcoder *TranscoderService + // strmResolve 把 strm 目标解析成最终直链。注入后 /Videos/{id}/stream 能直接 + // 302 到 CDN 地址,省掉 /api/strm/play 那一跳;未注入时保持原有的两跳行为, + // 因此测试与精简部署不受影响。 + strmResolve func(ctx context.Context, raw, userAgent string) (*StrmPlayResult, error) } // NewStreamService is the constructor. @@ -45,6 +49,16 @@ func NewStreamService(cfg *config.Config, log *zap.Logger, repo *repository.Cont } } +// SetStrmPlayTargetResolver 注入 strm 播放目标解析器(通常为 +// StrmService.ResolvePlayTargetWithUA)。注入后播放链路会在服务端完成换链并直接 +// 302 到最终直链,避免客户端在高延迟线路上多跟随一次 302。 +func (s *StreamService) SetStrmPlayTargetResolver(resolve func(ctx context.Context, raw, userAgent string) (*StrmPlayResult, error)) *StreamService { + if s != nil { + s.strmResolve = resolve + } + return s +} + // ErrMediaNotFound is returned when the media row or its file is missing. var ErrMediaNotFound = errors.New("media not found") diff --git a/internal/service/strm_play.go b/internal/service/strm_play.go index 7df1d3d..2b3952e 100644 --- a/internal/service/strm_play.go +++ b/internal/service/strm_play.go @@ -124,6 +124,16 @@ func (s *StrmService) resolveLocalPlay(ctx context.Context, rawPath string) (*St // - 绝对 http(s) 链接(直接透传,包含别的 MeBox / MediaStationGo 实例的播放端点) // - 其余协议(webdav:// 等)返回错误,由调用方决定是否静默跳过 func (s *StrmService) ResolvePlayTarget(ctx context.Context, raw string) (*StrmPlayResult, error) { + return s.ResolvePlayTargetWithUA(ctx, raw, "") +} + +// ResolvePlayTargetWithUA 与 ResolvePlayTarget 相同,但会把调用方的 User-Agent +// 透传给需要按 UA 换取直链的提供方(115 直链绑定换取时的 UA,换错会被 CDN 拒绝)。 +// +// 用途:播放链路在服务端直接把 strm 目标解析成最终直链并 302(见 +// StreamService.resolveDirectPlayTargetURL),此时必须带上播放器的 UA,才能拿到 +// 与 /api/strm/play 端点一致的、按 UA 分键缓存的那条直链。 +func (s *StrmService) ResolvePlayTargetWithUA(ctx context.Context, raw, userAgent string) (*StrmPlayResult, error) { raw = strings.TrimSpace(raw) if raw == "" { return nil, errors.New("空播放目标") @@ -150,14 +160,14 @@ func (s *StrmService) ResolvePlayTarget(ctx context.Context, raw string) (*StrmP if len(segs) < 1 || strings.TrimSpace(segs[0]) == "" { return nil, errors.New("无效的 strm 播放地址") } - return s.ResolvePlay(ctx, segs[0], u.Query()) + return s.ResolvePlay(ctx, segs[0], playbackQueryWithUA(u, userAgent)) case strings.HasPrefix(lowerPath, "/api/cloud/play/"): typ := strings.TrimSpace(strings.TrimPrefix(u.Path, "/api/cloud/play/")) acct, err := s.firstEnabledAccountOf(ctx, typ) if err != nil || acct == nil { return nil, errors.New("没有可用的网盘账号,无法解析直链") } - q := u.Query() + q := playbackQueryWithUA(u, userAgent) q.Set("acct", acct.ID) return s.ResolvePlay(ctx, typ, q) case u.Scheme == "http" || u.Scheme == "https": @@ -167,6 +177,19 @@ func (s *StrmService) ResolvePlayTarget(ctx context.Context, raw string) (*StrmP } } +// playbackQueryWithUA 复制播放目标的查询串并注入 __ua。ResolvePlay 的云盘提供方 +// 据此按播放器 UA 换取直链,与 /api/strm/play 端点写入 __ua 的语义保持一致。 +func playbackQueryWithUA(u *url.URL, userAgent string) url.Values { + q := url.Values{} + if u != nil { + q = url.Values(u.Query()) + } + if ua := strings.TrimSpace(userAgent); ua != "" { + q.Set("__ua", ua) + } + return q +} + // isLocalPlaybackTarget 报告播放地址是否属于本机。这里没有 HTTP 请求上下文, // 「本机」由 strm.base_url / 各同步目录覆盖的 base_url / 本机网盘账号共同界定 // (见 isInternalPlaybackTarget)。