mirror of
https://github.com/truewhile/MeBox.git
synced 2026-09-29 11:36:36 +08:00
fix: 资源占用/登录稳定性/QB整理入库/第三方播放404 综合修复
资源占用(Docker 部署 CPU/内存长期居高): - 云盘探测预算改为按尝试扣减,杜绝队列满时对每个文件反复入队 并刷出数万条 WARN(实测日志 41165 条) - 探测队列满时给文件挂 30 分钟退避 + 告警限速为每分钟一条 - 扫描时每个文件的海报/背景图由同步下载(单张最长 20s)改为 后台预取队列,云盘大库扫描不再串行拉图数小时 - PlaybackInfo 的云盘 ffprobe 探测改异步(原同步最长 8s, 既拖慢起播又放大云盘流量),带单飞去重 - 访问日志跳过 /api/health 与静态资源;logging.level/format 配置真正生效(此前是死配置) 登录稳定性(经常登录报错): - refresh token 未及时落库期间,刷新请求可识别「待落库令牌」, 不再把用户踢回登录页;轮换/登出后取消后台补写,防止旧令牌复活 QB 下载整理入库: - 新增 download.path_mappings 设置:自定义下载器→本程序路径映射 (每行 客户端路径=本地路径),并复用 compose 环境变量映射规则 - 应用重启后补整理最近 24h 内完成的种子(此前重启即永久漏掉) - 下载客户端初始化失败仍注册并惰性重连(容器启动顺序免疫) - 硬链接跨文件系统(EXDEV)自动降级为复制,保种语义不变 第三方播放器 404: - 播放处理器不再把所有错误吞成 404:媒体不存在→404, 云盘解析失败/STRM 关闭→502+原因 - 存库的云盘播放 URL 规范化为相对路径,免疫扫描时固化的旧 host - 云盘媒体 SupportsDirectPlay=false,强制走带鉴权的 DirectStream
This commit is contained in:
+16
-1
@@ -278,11 +278,26 @@ func isFrontendLibraryRoute(path string) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// newLogger 根据 cfg.Logging 构建 Zap。此前 logging.level / logging.format
|
||||
// 配置完全没有生效(固定 NewProduction),用户无法在生产环境降低日志量;
|
||||
// 配合每请求一条 INFO 访问日志,几小时即可产生几十 MB 日志,在 Docker
|
||||
// json-file 驱动下持续消耗磁盘 IO。
|
||||
func newLogger(cfg *config.Config) (*zap.Logger, error) {
|
||||
if cfg.App.Debug {
|
||||
return zap.NewDevelopment()
|
||||
}
|
||||
return zap.NewProduction()
|
||||
zapCfg := zap.NewProductionConfig()
|
||||
if level, err := zap.ParseAtomicLevel(strings.TrimSpace(cfg.Logging.Level)); err == nil && cfg.Logging.Level != "" {
|
||||
zapCfg.Level = level
|
||||
}
|
||||
if strings.EqualFold(strings.TrimSpace(cfg.Logging.Format), "console") {
|
||||
zapCfg.Encoding = "console"
|
||||
}
|
||||
if out := strings.TrimSpace(cfg.Logging.OutputPath); out != "" {
|
||||
zapCfg.OutputPaths = append(zapCfg.OutputPaths, out)
|
||||
zapCfg.ErrorOutputPaths = append(zapCfg.ErrorOutputPaths, out)
|
||||
}
|
||||
return zapCfg.Build()
|
||||
}
|
||||
|
||||
// getLocalIP returns the first non-loopback IPv4 address of the machine.
|
||||
|
||||
@@ -840,14 +840,27 @@ func embyVideoStreamHandler(svc *service.Container) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
uid := embyUserID(c)
|
||||
item, err := svc.Emby.Item(c.Request.Context(), c.Param("id"), uid)
|
||||
if err != nil || item == nil {
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if item == nil {
|
||||
c.Status(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
// 直接调用 Stream service 写入 response
|
||||
// 直接调用 Stream service 写入 response。
|
||||
// 此前这里把所有错误一律吞成 404:云盘 Cookie 过期、直链解析失败、
|
||||
// STRM 播放被关闭……在第三方播放器上全部表现为「404 不存在」,
|
||||
// 无法排查。现在区分:行不存在→404;云盘播放不可用/上游故障→502+原因。
|
||||
err = svc.Stream.ServeFile(c.Writer, c.Request, c.Param("id"))
|
||||
if err != nil {
|
||||
switch {
|
||||
case err == nil:
|
||||
case errors.Is(err, service.ErrMediaNotFound):
|
||||
c.Status(http.StatusNotFound)
|
||||
default:
|
||||
if !c.Writer.Written() {
|
||||
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -22,14 +22,26 @@ const (
|
||||
)
|
||||
|
||||
// RequestLogger logs one structured line per request.
|
||||
//
|
||||
// 健康检查与静态资源的成功请求被跳过:healthcheck 每 30s 一次、SPA 静态
|
||||
// 文件每页几十个请求,全部记 INFO 会让日志在几小时内膨胀到几十 MB,
|
||||
// 在 Docker json-file 日志驱动下白白消耗磁盘 IO。
|
||||
func RequestLogger(log *zap.Logger) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
start := time.Now()
|
||||
c.Next()
|
||||
path := c.Request.URL.Path
|
||||
status := c.Writer.Status()
|
||||
if status < 400 {
|
||||
if path == "/api/health" || strings.HasPrefix(path, "/assets/") ||
|
||||
path == "/favicon.ico" || path == "/favicon.svg" {
|
||||
return
|
||||
}
|
||||
}
|
||||
log.Info("http",
|
||||
zap.String("method", c.Request.Method),
|
||||
zap.String("path", c.Request.URL.Path),
|
||||
zap.Int("status", c.Writer.Status()),
|
||||
zap.String("path", path),
|
||||
zap.Int("status", status),
|
||||
zap.Duration("dur", time.Since(start)),
|
||||
zap.String("ip", c.ClientIP()),
|
||||
)
|
||||
|
||||
@@ -73,17 +73,20 @@ func (m *DownloadManager) LoadAll(ctx context.Context) error {
|
||||
}
|
||||
|
||||
if initErr := adapter.Initialize(ctx, cfg); initErr != nil {
|
||||
m.log.Warn("failed to initialize download client",
|
||||
// 初始化失败通常是 Docker 启动顺序问题(qBittorrent 还没就绪)。
|
||||
// 适配器内部支持按需重新登录(403/未登录时透明重试),所以
|
||||
// 这里仍然注册适配器,等下载器上线后自动恢复;此前直接 continue
|
||||
// 会让该客户端在应用重启前永久不可用,下载完成也无法整理入库。
|
||||
m.log.Warn("download client init failed; registered for lazy reconnect",
|
||||
zap.String("id", dc.ID),
|
||||
zap.String("name", dc.Name),
|
||||
zap.Error(initErr),
|
||||
)
|
||||
continue
|
||||
}
|
||||
|
||||
m.clients[dc.ID] = adapter
|
||||
m.configs[dc.ID] = cfg
|
||||
m.log.Info("download client initialized",
|
||||
m.log.Info("download client registered",
|
||||
zap.String("id", dc.ID),
|
||||
zap.String("name", dc.Name),
|
||||
zap.String("type", dc.Type),
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"math"
|
||||
"os"
|
||||
"net/url"
|
||||
"path"
|
||||
"path/filepath"
|
||||
@@ -814,7 +815,15 @@ func (d *DownloadService) processDownloadSnapshot(ctx context.Context, live []QB
|
||||
wasComplete, wasSeen := d.prevStates[stateKey]
|
||||
switch {
|
||||
case complete && (firstSnapshot || !wasSeen):
|
||||
// 首次快照里已完成的种子:此前一律标记「已见过」并跳过整理,
|
||||
// 导致「下载完成时应用恰好不在线/正在重启」的种子永远不会被
|
||||
// 自动整理入库。现在对最近完成的种子补一次整理
|
||||
// (onTorrentComplete 内部仍受 organize.auto 开关约束,且
|
||||
// 整理对已存在的目标文件幂等跳过)。
|
||||
d.prevStates[stateKey] = true
|
||||
if recentlyCompletedTorrent(torrent, time.Now()) {
|
||||
shouldQueue = true
|
||||
}
|
||||
case complete && !wasComplete:
|
||||
shouldQueue = true
|
||||
case complete:
|
||||
@@ -915,6 +924,20 @@ func (d *DownloadService) markCompletedTorrentOrganizeDone(torrent QBitTorrent)
|
||||
d.mu.Unlock()
|
||||
}
|
||||
|
||||
// completedTorrentCatchupWindow 限定重启补整理只覆盖最近完成的种子,
|
||||
// 防止每次启动都把全部历史种子重新过一遍整理流程。
|
||||
const completedTorrentCatchupWindow = 24 * time.Hour
|
||||
|
||||
// recentlyCompletedTorrent 报告该种子是否在补整理时间窗内完成。
|
||||
// qBittorrent 未提供 completion_on 时保守地返回 false。
|
||||
func recentlyCompletedTorrent(torrent QBitTorrent, now time.Time) bool {
|
||||
if torrent.CompletionOn <= 0 {
|
||||
return false
|
||||
}
|
||||
completed := time.Unix(torrent.CompletionOn, 0)
|
||||
return now.Sub(completed) <= completedTorrentCatchupWindow
|
||||
}
|
||||
|
||||
func completedTorrentQueueKey(torrent QBitTorrent) string {
|
||||
hash := strings.ToLower(strings.TrimSpace(torrent.Hash))
|
||||
if hash != "" {
|
||||
@@ -1010,7 +1033,7 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor
|
||||
d.log.Info("download completed, auto-organize disabled", zap.String("hash", torrent.Hash))
|
||||
return
|
||||
}
|
||||
source := d.completedTorrentSource(torrent)
|
||||
source := d.completedTorrentSource(ctx, torrent)
|
||||
if source == "" {
|
||||
d.log.Warn("download completed but payload path is not accessible",
|
||||
zap.String("hash", torrent.Hash),
|
||||
@@ -1045,13 +1068,23 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor
|
||||
zap.Int("errors", len(res.Errors)))
|
||||
}
|
||||
|
||||
func (d *DownloadService) completedTorrentSource(torrent QBitTorrent) string {
|
||||
// DownloadPathMappingsSettingKey 允许用户自定义「下载器路径 → 本程序路径」
|
||||
// 映射,每行一条,格式 `客户端路径=本地路径`(也接受 `=>` 或单个 `:` 分隔)。
|
||||
// qBittorrent 与本程序常在不同容器/主机里,对同一份数据看到的路径不同;
|
||||
// 此前映射表是写死的三条猜测,对不上时整理静默失败。
|
||||
const DownloadPathMappingsSettingKey = "download.path_mappings"
|
||||
|
||||
func (d *DownloadService) completedTorrentSource(ctx context.Context, torrent QBitTorrent) string {
|
||||
// 常见路径映射:qBittorrent容器路径 -> MediaStationGo容器路径
|
||||
mappings := map[string]string{
|
||||
"/var/apps/qBittorrent/shares/qBittorrent/Download": "/downloads",
|
||||
"/data/qBittorrent/downloads": "/downloads",
|
||||
"/downloads/qBittorrent": "/downloads",
|
||||
}
|
||||
// 用户自定义映射优先(可覆盖内置猜测)。
|
||||
for clientPrefix, localPrefix := range d.userPathMappings(ctx) {
|
||||
mappings[clientPrefix] = localPrefix
|
||||
}
|
||||
for _, candidate := range []string{
|
||||
torrent.ContentPath,
|
||||
filepath.Join(torrent.SavePath, torrent.Name),
|
||||
@@ -1064,6 +1097,55 @@ func (d *DownloadService) completedTorrentSource(torrent QBitTorrent) string {
|
||||
if translated := translateClientPath(clean, mappings); translated != "" {
|
||||
return translated
|
||||
}
|
||||
// 复用 compose 注入的 MEDIASTATION_DOWNLOAD_DIR/MEDIA_DIR 宿主机↔容器
|
||||
// 映射(与媒体库路径换算同一套规则),覆盖「qB 跑在宿主机、
|
||||
// 本程序在容器里」的最常见部署形态。
|
||||
for _, mapped := range mappedPathCandidates(clean) {
|
||||
if mapped == clean {
|
||||
continue
|
||||
}
|
||||
if _, err := os.Stat(mapped); err == nil {
|
||||
return mapped
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// userPathMappings 解析用户配置的下载器路径映射。
|
||||
func (d *DownloadService) userPathMappings(ctx context.Context) map[string]string {
|
||||
out := map[string]string{}
|
||||
if d == nil || d.repo == nil || d.repo.Setting == nil {
|
||||
return out
|
||||
}
|
||||
raw, err := d.repo.Setting.Get(ctx, DownloadPathMappingsSettingKey)
|
||||
if err != nil {
|
||||
return out
|
||||
}
|
||||
for _, line := range strings.Split(raw, "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" || strings.HasPrefix(line, "#") {
|
||||
continue
|
||||
}
|
||||
var from, to string
|
||||
switch {
|
||||
case strings.Contains(line, "=>"):
|
||||
parts := strings.SplitN(line, "=>", 2)
|
||||
from, to = parts[0], parts[1]
|
||||
case strings.Contains(line, "="):
|
||||
parts := strings.SplitN(line, "=", 2)
|
||||
from, to = parts[0], parts[1]
|
||||
case strings.Count(line, ":") == 1:
|
||||
parts := strings.SplitN(line, ":", 2)
|
||||
from, to = parts[0], parts[1]
|
||||
default:
|
||||
continue
|
||||
}
|
||||
from = strings.TrimSpace(from)
|
||||
to = strings.TrimSpace(to)
|
||||
if from != "" && to != "" {
|
||||
out[from] = to
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -136,7 +136,7 @@ func TestCompletedTorrentSourceDoesNotFallbackToSavePath(t *testing.T) {
|
||||
}
|
||||
svc := NewDownloadService(zap.NewNop(), newOrganizerTestRepo(t), NewHub(zap.NewNop()), nil)
|
||||
|
||||
got := svc.completedTorrentSource(QBitTorrent{
|
||||
got := svc.completedTorrentSource(t.Context(), QBitTorrent{
|
||||
Hash: "done123",
|
||||
Name: "Missing.Payload.S01",
|
||||
SavePath: savePath,
|
||||
@@ -190,6 +190,60 @@ func TestDownloadPollBaselinesAlreadyCompletedTorrents(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestDownloadPollCatchesUpRecentlyCompletedTorrents(t *testing.T) {
|
||||
repos := newOrganizerTestRepo(t)
|
||||
svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil)
|
||||
|
||||
svc.processDownloadSnapshot(t.Context(), []QBitTorrent{
|
||||
{Hash: "fresh-complete", Name: "Fresh Complete S01E01", Progress: 1, CompletionOn: time.Now().Add(-time.Hour).Unix()},
|
||||
{Hash: "stale-complete", Name: "Stale Complete S01E01", Progress: 1, CompletionOn: time.Now().Add(-48 * time.Hour).Unix()},
|
||||
{Hash: "no-timestamp", Name: "No Timestamp S01E01", Progress: 1},
|
||||
}, nil)
|
||||
|
||||
// 只有补整理时间窗内完成的种子会被补整理;无 completion_on 的保守跳过。
|
||||
if got := len(svc.organizeQueue); got != 1 {
|
||||
t.Fatalf("first poll queued %d organize jobs, want 1 (recent completion only)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompletedTorrentSourceUsesConfiguredMapping(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
localRoot := filepath.Join(root, "localdl")
|
||||
payload := filepath.Join(localRoot, "Show.S01")
|
||||
if err := os.MkdirAll(payload, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
repos := newOrganizerTestRepo(t)
|
||||
if err := repos.Setting.Set(t.Context(), DownloadPathMappingsSettingKey, "/qb/downloads="+localRoot); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil)
|
||||
|
||||
got := svc.completedTorrentSource(t.Context(), QBitTorrent{ContentPath: "/qb/downloads/Show.S01"})
|
||||
if got != payload {
|
||||
t.Fatalf("completedTorrentSource = %q, want %q", got, payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUserPathMappingsParsing(t *testing.T) {
|
||||
repos := newOrganizerTestRepo(t)
|
||||
raw := "# comment\n/a=/b\n/c => /d\n/e:/f\nbad-line\n"
|
||||
if err := repos.Setting.Set(t.Context(), DownloadPathMappingsSettingKey, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil)
|
||||
got := svc.userPathMappings(t.Context())
|
||||
want := map[string]string{"/a": "/b", "/c": "/d", "/e": "/f"}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("userPathMappings = %v, want %v", got, want)
|
||||
}
|
||||
for k, v := range want {
|
||||
if got[k] != v {
|
||||
t.Fatalf("mapping %q = %q, want %q", k, got[k], v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPublicDownloadTitleUsesMagnetDisplayName(t *testing.T) {
|
||||
got := publicDownloadTitle("magnet:?xt=urn:btih:abc&dn=%E6%B5%8B%E8%AF%95%E5%BD%B1%E7%89%87")
|
||||
if got != "测试影片" {
|
||||
|
||||
@@ -69,6 +69,9 @@ type EmbyService struct {
|
||||
|
||||
visibilityMu sync.RWMutex
|
||||
visibilityCache map[string]embyVisibilityCacheEntry
|
||||
|
||||
cloudProbeMu sync.Mutex
|
||||
cloudProbeInFlight map[string]struct{}
|
||||
}
|
||||
|
||||
type cloudPlaybackResolver interface {
|
||||
@@ -1476,6 +1479,13 @@ func (e *EmbyService) PlaybackInfo(ctx context.Context, mediaID, userID string)
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ensureCloudTrackMetadata 在后台补齐云盘媒体的轨道元数据。
|
||||
//
|
||||
// 注意必须是异步的:此前这里在 PlaybackInfo 请求路径上同步执行
|
||||
// CloudResolve + ffprobe(HTTP)(最长 8 秒),既把第三方播放器的起播时间
|
||||
// 拖长到秒级,又让每一次点开详情/起播都可能触发一次云盘数据下载,是
|
||||
// Docker 部署下 CPU/带宽长期居高的来源之一。探测结果落库后,下一次
|
||||
// 请求自然能读到完整元数据。
|
||||
func (e *EmbyService) ensureCloudTrackMetadata(ctx context.Context, m *model.Media) {
|
||||
if e == nil || m == nil || e.storage == nil || e.probe == nil || !mediaTrackMetadataMissing(m) {
|
||||
return
|
||||
@@ -1484,30 +1494,48 @@ func (e *EmbyService) ensureCloudTrackMetadata(ctx context.Context, m *model.Med
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
probeCtx, cancel := context.WithTimeout(ctx, 8*time.Second)
|
||||
defer cancel()
|
||||
link, err := e.storage.CloudResolve(probeCtx, typ, ref, "")
|
||||
if err != nil {
|
||||
if e.log != nil {
|
||||
e.log.Debug("resolve cloud media for playback probe failed", zap.String("media_id", m.ID), zap.Error(err))
|
||||
mediaID := m.ID
|
||||
e.cloudProbeMu.Lock()
|
||||
if e.cloudProbeInFlight == nil {
|
||||
e.cloudProbeInFlight = make(map[string]struct{})
|
||||
}
|
||||
if _, busy := e.cloudProbeInFlight[mediaID]; busy {
|
||||
e.cloudProbeMu.Unlock()
|
||||
return
|
||||
}
|
||||
e.cloudProbeInFlight[mediaID] = struct{}{}
|
||||
e.cloudProbeMu.Unlock()
|
||||
|
||||
go func() {
|
||||
defer func() {
|
||||
e.cloudProbeMu.Lock()
|
||||
delete(e.cloudProbeInFlight, mediaID)
|
||||
e.cloudProbeMu.Unlock()
|
||||
}()
|
||||
probeCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
link, err := e.storage.CloudResolve(probeCtx, typ, ref, "")
|
||||
if err != nil {
|
||||
if e.log != nil {
|
||||
e.log.Debug("resolve cloud media for playback probe failed", zap.String("media_id", mediaID), zap.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
probe, err := e.probe.ProbeHTTP(probeCtx, link.URL, link.Headers)
|
||||
if err != nil {
|
||||
if e.log != nil {
|
||||
e.log.Debug("playback cloud ffprobe failed", zap.String("media_id", m.ID), zap.Error(err))
|
||||
probe, err := e.probe.ProbeHTTP(probeCtx, link.URL, link.Headers)
|
||||
if err != nil {
|
||||
if e.log != nil {
|
||||
e.log.Debug("playback cloud ffprobe failed", zap.String("media_id", mediaID), zap.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
updates := probeResultUpdates(probe)
|
||||
if len(updates) == 0 {
|
||||
return
|
||||
}
|
||||
if err := e.repo.DB.WithContext(ctx).Model(&model.Media{}).Where("id = ?", m.ID).Updates(updates).Error; err != nil && e.log != nil {
|
||||
e.log.Debug("persist playback cloud probe failed", zap.String("media_id", m.ID), zap.Error(err))
|
||||
}
|
||||
applyProbeResultToMediaValue(m, probe)
|
||||
updates := probeResultUpdates(probe)
|
||||
if len(updates) == 0 {
|
||||
return
|
||||
}
|
||||
if err := e.repo.DB.WithContext(probeCtx).Model(&model.Media{}).Where("id = ?", mediaID).Updates(updates).Error; err != nil && e.log != nil {
|
||||
e.log.Debug("persist playback cloud probe failed", zap.String("media_id", mediaID), zap.Error(err))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func mediaTrackMetadataMissing(m *model.Media) bool {
|
||||
@@ -1630,11 +1658,16 @@ func (e *EmbyService) mediaSource(m *model.Media, asEmbedded, directOnly bool) m
|
||||
"RequiresClosing": false,
|
||||
"ReadAtNativeFramerate": false,
|
||||
"SupportsTranscoding": !directOnly,
|
||||
"SupportsDirectStream": true,
|
||||
"SupportsDirectPlay": true,
|
||||
"SupportsProbing": true,
|
||||
"RunTimeTicks": int64(m.DurationSec) * 10_000_000,
|
||||
"MediaStreams": e.mediaStreams(m),
|
||||
// 云盘媒体禁用 DirectPlay:DirectPlay 语义是「客户端直接访问
|
||||
// Path」,而云盘媒体的 Path 是不带鉴权 token 的内部 /api/cloud/play
|
||||
// 路径,Infuse/VidHub 等播放器直接请求会得到 401/404。强制它们走
|
||||
// DirectStream(/Videos/{id}/stream?api_key=...),由服务端校验后
|
||||
// 302 到云盘直链。
|
||||
"SupportsDirectStream": true,
|
||||
"SupportsDirectPlay": !isCloud,
|
||||
"SupportsProbing": true,
|
||||
"RunTimeTicks": int64(m.DurationSec) * 10_000_000,
|
||||
"MediaStreams": e.mediaStreams(m),
|
||||
}
|
||||
if !asEmbedded {
|
||||
src["DirectStreamUrl"] = embyDirectStreamURL(m.ID, container)
|
||||
|
||||
@@ -3,6 +3,7 @@ package service
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"go.uber.org/zap"
|
||||
@@ -405,30 +406,45 @@ func TestEmbyPlaybackInfoProbesMissingCloudTrackMetadata(t *testing.T) {
|
||||
}
|
||||
svc.SetCloudProbe(resolver, prober)
|
||||
|
||||
pb, err := svc.PlaybackInfo(t.Context(), "cloud-probe-1", "user-1")
|
||||
if err != nil {
|
||||
if _, err := svc.PlaybackInfo(t.Context(), "cloud-probe-1", "user-1"); err != nil {
|
||||
t.Fatalf("playback info: %v", err)
|
||||
}
|
||||
|
||||
// 探测现在是异步的(同步探测曾把起播拖慢最多 8 秒并放大云盘流量)。
|
||||
// 轮询等待后台探测结果落库。
|
||||
var persisted model.Media
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for {
|
||||
if err := svc.repo.DB.First(&persisted, "id = ?", "cloud-probe-1").Error; err != nil {
|
||||
t.Fatalf("reload media: %v", err)
|
||||
}
|
||||
if persisted.DurationSec > 0 || time.Now().After(deadline) {
|
||||
break
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
if persisted.DurationSec != 3661 || persisted.Width != 3840 || persisted.Height != 2160 || persisted.VideoCodec != "hevc" || persisted.AudioCodec != "eac3" {
|
||||
t.Fatalf("probe metadata not persisted: %#v", persisted)
|
||||
}
|
||||
if resolver.typ != "openlist" || resolver.ref != "/Movies/Movie.mkv" {
|
||||
t.Fatalf("resolver called with typ=%q ref=%q", resolver.typ, resolver.ref)
|
||||
}
|
||||
if prober.rawURL != "http://cdn.example.test/Movie.mkv" || prober.headers["Authorization"] != "Bearer probe-token" {
|
||||
t.Fatalf("probe called with url=%q headers=%#v", prober.rawURL, prober.headers)
|
||||
}
|
||||
|
||||
// 落库之后,再次请求 PlaybackInfo 应当带上完整轨道元数据。
|
||||
pb, err := svc.PlaybackInfo(t.Context(), "cloud-probe-1", "user-1")
|
||||
if err != nil {
|
||||
t.Fatalf("playback info (second): %v", err)
|
||||
}
|
||||
src := pb["MediaSources"].([]map[string]any)[0]
|
||||
if src["RunTimeTicks"] != int64(3661)*10_000_000 {
|
||||
t.Fatalf("runtime ticks not filled from probe: %#v", src)
|
||||
t.Fatalf("runtime ticks not filled after async probe: %#v", src)
|
||||
}
|
||||
streams := src["MediaStreams"].([]map[string]any)
|
||||
if len(streams) != 2 || streams[0]["Codec"] != "hevc" || streams[1]["Codec"] != "eac3" {
|
||||
t.Fatalf("media streams not filled from probe: %#v", streams)
|
||||
}
|
||||
var persisted model.Media
|
||||
if err := svc.repo.DB.First(&persisted, "id = ?", "cloud-probe-1").Error; err != nil {
|
||||
t.Fatalf("reload media: %v", err)
|
||||
}
|
||||
if persisted.DurationSec != 3661 || persisted.Width != 3840 || persisted.Height != 2160 || persisted.VideoCodec != "hevc" || persisted.AudioCodec != "eac3" {
|
||||
t.Fatalf("probe metadata not persisted: %#v", persisted)
|
||||
t.Fatalf("media streams not filled after async probe: %#v", streams)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -438,6 +454,11 @@ func newTestEmbyService(t *testing.T) *EmbyService {
|
||||
if err != nil {
|
||||
t.Fatalf("open db: %v", err)
|
||||
}
|
||||
// 内存库 + 异步探测协程:限制为单连接,避免连接池新建连接时
|
||||
// 拿到一个空白的 :memory: 实例(no such table)。
|
||||
if sqlDB, err := db.DB(); err == nil {
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
}
|
||||
if err := db.AutoMigrate(&model.Library{}, &model.Series{}, &model.Media{}, &model.Favorite{}, &model.PlaybackHistory{}, &model.User{}, &model.Setting{}); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
|
||||
@@ -59,6 +59,10 @@ type QBitTorrent struct {
|
||||
// root folder. Prefer it for automatic organize so we do not scan the whole
|
||||
// download category.
|
||||
ContentPath string `json:"content_path"`
|
||||
// CompletionOn 是 qBittorrent 报告的完成时间(Unix 秒,未完成为 0 或负值)。
|
||||
// 用于应用重启后的「补整理」判断:只补最近完成的种子,避免每次启动
|
||||
// 都重新触发全部历史种子的整理。
|
||||
CompletionOn int64 `json:"completion_on"`
|
||||
}
|
||||
|
||||
// QBitClient is a thread-safe qBittorrent v2 API client.
|
||||
|
||||
@@ -71,6 +71,8 @@ type ScannerService struct {
|
||||
cloudMediaProbeMu sync.Mutex
|
||||
cloudMediaProbing map[string]struct{}
|
||||
cloudMediaProbeBackoff map[string]time.Time
|
||||
cloudMediaProbeWarnMu sync.Mutex
|
||||
cloudMediaProbeLastWarn time.Time
|
||||
}
|
||||
|
||||
// NewScannerService is the constructor.
|
||||
@@ -269,6 +271,10 @@ const maxCloudMediaProbeQueuePerScan = 32
|
||||
|
||||
const cloudMediaProbeFailureBackoff = 6 * time.Hour
|
||||
|
||||
// cloudMediaProbeQueueFullBackoff 是探测队列饱和时给单个文件挂的短退避,
|
||||
// 防止后续扫描轮次对同一批文件反复尝试入队。
|
||||
const cloudMediaProbeQueueFullBackoff = 30 * time.Minute
|
||||
|
||||
// CloudScanStatus is the operator-facing state for long-running cloud scans.
|
||||
type CloudScanStatus struct {
|
||||
LibraryID string `json:"library_id"`
|
||||
@@ -359,9 +365,29 @@ func (s *ScannerService) queueCloudMediaProbe(typ, ref, path string) bool {
|
||||
default:
|
||||
s.cloudMediaProbeMu.Lock()
|
||||
delete(s.cloudMediaProbing, path)
|
||||
// 队列满说明探测工人已饱和;给该文件挂一个短退避,避免下一轮
|
||||
// 扫描立刻重复尝试同一批文件。
|
||||
if s.cloudMediaProbeBackoff == nil {
|
||||
s.cloudMediaProbeBackoff = make(map[string]time.Time)
|
||||
}
|
||||
s.cloudMediaProbeBackoff[path] = time.Now().Add(cloudMediaProbeQueueFullBackoff)
|
||||
s.cloudMediaProbeMu.Unlock()
|
||||
if s.log != nil {
|
||||
s.log.Warn("cloud media probe queue full", zap.String("provider", typ), zap.String("path", path))
|
||||
// 限速告警:队列满在大库扫描中是常态而非异常,逐条 WARN 会
|
||||
// 在几小时内产生数万行日志(真实环境出现过 41165 条)。
|
||||
now := time.Now()
|
||||
s.cloudMediaProbeWarnMu.Lock()
|
||||
shouldWarn := now.Sub(s.cloudMediaProbeLastWarn) >= time.Minute
|
||||
if shouldWarn {
|
||||
s.cloudMediaProbeLastWarn = now
|
||||
}
|
||||
s.cloudMediaProbeWarnMu.Unlock()
|
||||
if shouldWarn {
|
||||
s.log.Warn("cloud media probe queue full; deferring remaining probes (logged at most once per minute)",
|
||||
zap.String("provider", typ), zap.String("path", path))
|
||||
} else {
|
||||
s.log.Debug("cloud media probe queue full", zap.String("provider", typ), zap.String("path", path))
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -372,14 +398,13 @@ func (s *ScannerService) queueCloudMediaProbeWithBudget(typ, ref, path string, b
|
||||
if *budget <= 0 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
if !s.queueCloudMediaProbe(typ, ref, path) {
|
||||
return false
|
||||
}
|
||||
if budget != nil {
|
||||
// 预算按「尝试」扣减而不是按「成功入队」扣减。否则当探测队列被
|
||||
// 其他扫描填满时,本次扫描会对剩下的每一个文件都尝试入队并各打
|
||||
// 一条日志——真实环境里曾因此产生过 4 万多条 "queue full" WARN,
|
||||
// 这本身就是一笔可观的 CPU/磁盘开销。
|
||||
*budget--
|
||||
}
|
||||
return true
|
||||
return s.queueCloudMediaProbe(typ, ref, path)
|
||||
}
|
||||
|
||||
func (s *ScannerService) beginCloudScan(ctx context.Context, lib *model.Library, mount CloudMountInfo) (context.Context, func(*ScanResult, error), error) {
|
||||
@@ -917,7 +942,13 @@ func (s *ScannerService) scanCloudLibrary(ctx context.Context, lib *model.Librar
|
||||
displayPath := joinCloudDisplayPath(displayDir, entry.Name)
|
||||
path := cloudMediaPath(typ, displayPath)
|
||||
localMeta := s.cloudFileMetadata(ctx, typ, displayPath, entry.Name, sidecars, dirMeta, librarySupportsSeasons(lib))
|
||||
s.cacheCloudMetadataArtworkNow(ctx, localMeta)
|
||||
// 每个文件的海报/背景图改走后台预取队列。此前这里是同步
|
||||
// CloudResolve+下载(每张最多 20s 超时),几千个文件的云盘库
|
||||
// 扫描会变成持续数小时的串行下载,把 CPU/带宽长期吃满。
|
||||
if localMeta != nil {
|
||||
s.queueCloudArtworkPrefetch(localMeta.PosterURL)
|
||||
s.queueCloudArtworkPrefetch(localMeta.BackdropURL)
|
||||
}
|
||||
candidate := cloudCandidate{
|
||||
ref: ref,
|
||||
name: entry.Name,
|
||||
|
||||
@@ -57,6 +57,32 @@ func NewStreamService(cfg *config.Config, log *zap.Logger, repo *repository.Cont
|
||||
// ErrMediaNotFound is returned when the media row or its file is missing.
|
||||
var ErrMediaNotFound = errors.New("media not found")
|
||||
|
||||
// ErrCloudPlaybackUnavailable 表示媒体行存在但属于云盘媒体、且当前无法
|
||||
// 构造可用的播放重定向(例如 STRM 播放被关闭或 STRMURL 缺失)。调用方
|
||||
// 应把它与「媒体不存在」区分开,避免把配置类故障当成 404 返回给播放器。
|
||||
var ErrCloudPlaybackUnavailable = errors.New("cloud media playback unavailable: strm playback disabled or media missing play url; re-scan the library or enable strm playback")
|
||||
|
||||
// normalizeCloudPlayTarget 把存库的云盘播放 URL 规范化为相对路径。
|
||||
//
|
||||
// STRMURL 是扫描时根据当时的 server_url/请求地址生成并固化进数据库的。
|
||||
// 在 Windows 开发机上扫描、再部署到 Docker(或更换了内网 IP/域名)后,
|
||||
// 这些绝对 URL 会指向已失效的旧地址,第三方播放器跟随 302 就会拿到
|
||||
// 连接失败/404。这里只要能从 URL 中解析出 provider+ref,就重建为相对
|
||||
// /api/cloud/play 路径,由 absoluteInternalRedirect 基于「当前请求」补全
|
||||
// host,从而对历史脏数据免疫。
|
||||
func normalizeCloudPlayTarget(raw string) string {
|
||||
typ, ref, ok := parseCloudMediaPlaybackURL(raw)
|
||||
if !ok {
|
||||
return raw
|
||||
}
|
||||
return BuildRelativeCloudPlayURL(typ, ref)
|
||||
}
|
||||
|
||||
// BuildRelativeCloudPlayURL 构造相对的云盘播放 API 路径。
|
||||
func BuildRelativeCloudPlayURL(typ, ref string) string {
|
||||
return "/api/cloud/play/" + url.PathEscape(strings.TrimSpace(typ)) + "?" + url.Values{"ref": []string{ref}}.Encode()
|
||||
}
|
||||
|
||||
// directPlayOnly reports whether the admin enabled「客户端直连解码」mode,
|
||||
// in which the host never transcodes (HLS is refused) and all playback is
|
||||
// handled by the client (direct play / 302 redirect).
|
||||
@@ -209,10 +235,18 @@ func (s *StreamService) ServeFile(w http.ResponseWriter, r *http.Request, mediaI
|
||||
return ErrMediaNotFound
|
||||
}
|
||||
if strings.TrimSpace(m.STRMURL) != "" && STRMPlaybackEnabled(r.Context(), s.repo) {
|
||||
target := withAuthTokenForInternalRedirect(m.STRMURL, r, PublicServerURL(r.Context(), s.repo, s.cfg))
|
||||
// 云盘播放 URL 先规范化为相对路径,免疫扫描时固化的旧 host。
|
||||
target := normalizeCloudPlayTarget(m.STRMURL)
|
||||
target = withAuthTokenForInternalRedirect(target, r, PublicServerURL(r.Context(), s.repo, s.cfg))
|
||||
http.Redirect(w, r, absoluteInternalRedirect(target, r), http.StatusFound)
|
||||
return nil
|
||||
}
|
||||
if strings.HasPrefix(strings.ToLower(strings.TrimSpace(m.Path)), "cloud://") {
|
||||
// 云盘媒体没有本地文件可回退;走到这里说明 STRM 播放被关闭或
|
||||
// STRMURL 缺失。返回明确错误而不是笼统的「文件不存在」,
|
||||
// 处理器据此回 502 + 原因,方便用户在播放器/日志里定位。
|
||||
return ErrCloudPlaybackUnavailable
|
||||
}
|
||||
f, err := os.Open(m.Path)
|
||||
if err != nil {
|
||||
return ErrMediaNotFound
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestNormalizeCloudPlayTarget 验证存库的云盘播放 URL(可能携带扫描时的
|
||||
// 旧 host)被规范化为相对路径,使 302 始终基于当前请求地址构造。
|
||||
func TestNormalizeCloudPlayTarget(t *testing.T) {
|
||||
ref := "/电影/某部影片 (2024)/movie.mkv"
|
||||
stale := "http://192.168.1.4:9011/api/cloud/play/openlist?ref=" + url.QueryEscape(ref)
|
||||
got := normalizeCloudPlayTarget(stale)
|
||||
want := BuildRelativeCloudPlayURL("openlist", ref)
|
||||
if got != want {
|
||||
t.Fatalf("normalizeCloudPlayTarget = %q, want %q", got, want)
|
||||
}
|
||||
parsed, err := url.Parse(got)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if parsed.IsAbs() || parsed.Host != "" {
|
||||
t.Fatalf("normalized target should be relative, got %q", got)
|
||||
}
|
||||
if parsed.Query().Get("ref") != ref {
|
||||
t.Fatalf("ref round-trip failed: %q", parsed.Query().Get("ref"))
|
||||
}
|
||||
|
||||
// 非云盘播放 URL 保持原样(WebDAV/直链等)。
|
||||
passthrough := "https://dav.example.com/media/file.mkv"
|
||||
if got := normalizeCloudPlayTarget(passthrough); got != passthrough {
|
||||
t.Fatalf("non-cloud target should pass through, got %q", got)
|
||||
}
|
||||
}
|
||||
@@ -103,7 +103,9 @@ func TestServeFileHonorsSTRMPlaybackDisabled(t *testing.T) {
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
err := svc.ServeFile(w, req, "cloud-1")
|
||||
if err != ErrMediaNotFound {
|
||||
// 云盘媒体在 STRM 播放关闭时返回明确的「云盘播放不可用」错误,
|
||||
// 而不是和「媒体不存在」混在一起(后者会让播放器显示 404)。
|
||||
if err != ErrCloudPlaybackUnavailable {
|
||||
t.Fatalf("disabled STRM should not redirect cloud media, err=%v status=%d location=%q", err, w.Code, w.Header().Get("Location"))
|
||||
}
|
||||
if loc := w.Header().Get("Location"); loc != "" {
|
||||
|
||||
@@ -42,12 +42,22 @@ type TokenService struct {
|
||||
log *zap.Logger
|
||||
repo *repository.Container
|
||||
delayedStoreMu sync.Mutex
|
||||
delayedStores map[string]struct{}
|
||||
// delayedStores 记录「已发给客户端但还没写进库」的 refresh token。
|
||||
// 键是 token 哈希;值携带签发信息,让 Refresh 在落库完成前也能识别
|
||||
// 这些令牌——否则用户登录成功、一小时后 access token 过期,刷新时
|
||||
// 因为 refresh token 从未落库而被判定无效,被强制踢回登录页,
|
||||
// 表现就是「经常登录报错」。
|
||||
delayedStores map[string]pendingRefreshToken
|
||||
}
|
||||
|
||||
type pendingRefreshToken struct {
|
||||
UserID string
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
|
||||
// NewTokenService 创建令牌服务实例。
|
||||
func NewTokenService(cfg *config.Config, log *zap.Logger, repo *repository.Container) *TokenService {
|
||||
return &TokenService{cfg: cfg, log: log, repo: repo, delayedStores: make(map[string]struct{})}
|
||||
return &TokenService{cfg: cfg, log: log, repo: repo, delayedStores: make(map[string]pendingRefreshToken)}
|
||||
}
|
||||
|
||||
// TokenPair 包含访问令牌和刷新令牌。
|
||||
@@ -113,7 +123,7 @@ func (s *TokenService) issuePair(ctx context.Context, userID, role, tier string,
|
||||
zap.String("user_id", userID),
|
||||
zap.Error(err))
|
||||
}
|
||||
if s.trackDelayedStore(userID, tokenHash) {
|
||||
if s.trackDelayedStore(userID, tokenHash, rt.ExpiresAt) {
|
||||
go s.storeRefreshTokenEventually(userID, tokenHash, rt.ExpiresAt)
|
||||
}
|
||||
}
|
||||
@@ -142,6 +152,11 @@ func (s *TokenService) storeRefreshTokenEventually(userID, tokenHash string, exp
|
||||
for attempt := 1; attempt <= 8; attempt++ {
|
||||
timer := time.NewTimer(delay)
|
||||
<-timer.C
|
||||
// 令牌可能已在等待期间被轮换/登出(从 pending 表移除),
|
||||
// 此时绝不能再写库,否则会复活一个已被替换的旧令牌。
|
||||
if _, stillPending := s.pendingDelayedStore(tokenHash); !stillPending {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
err := s.storeRefreshToken(ctx, &model.RefreshToken{
|
||||
UserID: userID,
|
||||
@@ -173,20 +188,19 @@ func (s *TokenService) storeRefreshTokenEventually(userID, tokenHash string, exp
|
||||
}
|
||||
}
|
||||
|
||||
func (s *TokenService) trackDelayedStore(userID, tokenHash string) bool {
|
||||
func (s *TokenService) trackDelayedStore(userID, tokenHash string, expiresAt time.Time) bool {
|
||||
if s == nil {
|
||||
return false
|
||||
}
|
||||
key := userID + "\x00" + tokenHash
|
||||
s.delayedStoreMu.Lock()
|
||||
defer s.delayedStoreMu.Unlock()
|
||||
if s.delayedStores == nil {
|
||||
s.delayedStores = make(map[string]struct{})
|
||||
s.delayedStores = make(map[string]pendingRefreshToken)
|
||||
}
|
||||
if _, ok := s.delayedStores[key]; ok {
|
||||
if _, ok := s.delayedStores[tokenHash]; ok {
|
||||
return false
|
||||
}
|
||||
s.delayedStores[key] = struct{}{}
|
||||
s.delayedStores[tokenHash] = pendingRefreshToken{UserID: userID, ExpiresAt: expiresAt}
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -194,12 +208,22 @@ func (s *TokenService) untrackDelayedStore(userID, tokenHash string) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
key := userID + "\x00" + tokenHash
|
||||
s.delayedStoreMu.Lock()
|
||||
delete(s.delayedStores, key)
|
||||
delete(s.delayedStores, tokenHash)
|
||||
s.delayedStoreMu.Unlock()
|
||||
}
|
||||
|
||||
// pendingDelayedStore 返回尚未落库的 refresh token 信息(如果存在)。
|
||||
func (s *TokenService) pendingDelayedStore(tokenHash string) (pendingRefreshToken, bool) {
|
||||
if s == nil {
|
||||
return pendingRefreshToken{}, false
|
||||
}
|
||||
s.delayedStoreMu.Lock()
|
||||
defer s.delayedStoreMu.Unlock()
|
||||
pending, ok := s.delayedStores[tokenHash]
|
||||
return pending, ok
|
||||
}
|
||||
|
||||
func (s *TokenService) maxActiveRefreshTokens(ctx context.Context) int {
|
||||
cfg := loadBotConfig(ctx, s.repo)
|
||||
if cfg.MaxLoggedClients < 1 {
|
||||
@@ -244,7 +268,17 @@ func (s *TokenService) Refresh(ctx context.Context, refreshToken string) (*Token
|
||||
return nil, err
|
||||
}
|
||||
if rt == nil {
|
||||
return nil, ErrInvalidRefreshToken
|
||||
// 登录高峰/扫描写压力下,refresh token 可能还在后台补写队列里
|
||||
// 没来得及落库。此时令牌对客户端而言是合法的,不能判无效。
|
||||
pending, ok := s.pendingDelayedStore(tokenHash)
|
||||
if !ok || time.Now().After(pending.ExpiresAt) {
|
||||
return nil, ErrInvalidRefreshToken
|
||||
}
|
||||
rt = &model.RefreshToken{
|
||||
UserID: pending.UserID,
|
||||
TokenHash: tokenHash,
|
||||
ExpiresAt: pending.ExpiresAt,
|
||||
}
|
||||
}
|
||||
|
||||
// 检查是否已撤销
|
||||
@@ -272,10 +306,11 @@ func (s *TokenService) Refresh(ctx context.Context, refreshToken string) (*Token
|
||||
return nil, ErrUserExpired
|
||||
}
|
||||
|
||||
// 撤销旧的 Refresh Token
|
||||
// 撤销旧的 Refresh Token(包括可能仍在后台补写队列里的副本)。
|
||||
if err := s.repo.RefreshToken.Revoke(ctx, tokenHash); err != nil {
|
||||
s.log.Warn("failed to revoke old refresh token", zap.Error(err))
|
||||
}
|
||||
s.untrackDelayedStore(rt.UserID, tokenHash)
|
||||
|
||||
// 签发新的令牌对
|
||||
return s.IssuePair(ctx, user.ID, user.Role, user.Tier)
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"go.uber.org/zap"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/ShukeBta/MediaStationGo/internal/config"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/model"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/repository"
|
||||
)
|
||||
|
||||
func newTokenTestRepo(t *testing.T) *repository.Container {
|
||||
t.Helper()
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.AutoMigrate(&model.User{}, &model.RefreshToken{}, &model.Setting{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return repository.New(db)
|
||||
}
|
||||
|
||||
// TestRefreshAcceptsPendingDelayedToken 验证:登录时因 SQLite 写压力未及时
|
||||
// 落库的 refresh token(仍在后台补写队列中)在刷新时被接受,而不是把用户
|
||||
// 踢回登录页(历史上「经常登录报错」的来源之一)。
|
||||
func TestRefreshAcceptsPendingDelayedToken(t *testing.T) {
|
||||
repos := newTokenTestRepo(t)
|
||||
cfg := &config.Config{}
|
||||
cfg.Secrets.JWTSecret = "test-secret"
|
||||
svc := NewTokenService(cfg, zap.NewNop(), repos)
|
||||
|
||||
u := &model.User{Username: "u1", PasswordHash: "x", Role: "user", Tier: "free", IsActive: true}
|
||||
if err := repos.User.Create(t.Context(), u); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
refreshToken := "pending-token-value"
|
||||
hash := repository.HashToken(refreshToken)
|
||||
if !svc.trackDelayedStore(u.ID, hash, time.Now().Add(time.Hour)) {
|
||||
t.Fatal("trackDelayedStore returned false")
|
||||
}
|
||||
|
||||
pair, err := svc.Refresh(t.Context(), refreshToken)
|
||||
if err != nil {
|
||||
t.Fatalf("Refresh rejected pending delayed token: %v", err)
|
||||
}
|
||||
if pair == nil || pair.AccessToken == "" || pair.RefreshToken == "" {
|
||||
t.Fatalf("Refresh returned incomplete pair: %+v", pair)
|
||||
}
|
||||
// 轮换后旧令牌应从 pending 表移除,不能再次使用。
|
||||
if _, still := svc.pendingDelayedStore(hash); still {
|
||||
t.Fatal("rotated pending token still tracked")
|
||||
}
|
||||
if _, err := svc.Refresh(t.Context(), refreshToken); err == nil {
|
||||
t.Fatal("rotated pending token should not refresh twice")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefreshRejectsExpiredPendingToken 验证过期的待落库令牌不会被接受。
|
||||
func TestRefreshRejectsExpiredPendingToken(t *testing.T) {
|
||||
repos := newTokenTestRepo(t)
|
||||
cfg := &config.Config{}
|
||||
cfg.Secrets.JWTSecret = "test-secret"
|
||||
svc := NewTokenService(cfg, zap.NewNop(), repos)
|
||||
|
||||
refreshToken := "expired-pending"
|
||||
hash := repository.HashToken(refreshToken)
|
||||
svc.trackDelayedStore("user-x", hash, time.Now().Add(-time.Minute))
|
||||
|
||||
if _, err := svc.Refresh(t.Context(), refreshToken); err == nil {
|
||||
t.Fatal("expired pending token should be rejected")
|
||||
}
|
||||
}
|
||||
@@ -63,6 +63,13 @@ func transferFile(src, dst string, mode TransferMode) error {
|
||||
return copyFile(src, dst)
|
||||
case TransferHardlink:
|
||||
if err := os.Link(src, dst); err != nil {
|
||||
// Docker 部署里下载目录和媒体目录往往是两个独立的 bind mount,
|
||||
// 即使在宿主机上同属一块盘,容器内 os.Link 也会因跨文件系统
|
||||
// (EXDEV) 失败。此前直接报错导致 PT 下载完成后整理静默中断;
|
||||
// 现在自动降级为复制(保留源文件继续做种,语义一致)。
|
||||
if copyErr := copyFile(src, dst); copyErr == nil {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("hardlink failed: %w; source and target must be on the same filesystem, choose copy if you want to duplicate data", err)
|
||||
}
|
||||
return nil
|
||||
|
||||
Reference in New Issue
Block a user