From d270e9034f879907723a10c7c4f847c5f433e377 Mon Sep 17 00:00:00 2001 From: ShukeBta Date: Fri, 12 Jun 2026 19:06:47 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=98=B2=E6=AD=A2=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E6=95=B4=E7=90=86=E6=89=AB=E6=8F=8F=E9=98=BB=E5=A1=9E=E7=99=BB?= =?UTF-8?q?=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/handler/auth.go | 2 +- internal/service/audit.go | 14 ++++ internal/service/auth.go | 15 ++++- internal/service/auth_user_limits_test.go | 5 ++ internal/service/downloads.go | 58 +++++++++++++++- internal/service/downloads_test.go | 69 +++++++++++++++++++ internal/service/organizer_scan.go | 8 +++ internal/service/path_translation.go | 11 +++- internal/service/scheduler.go | 18 +++-- internal/service/service.go | 66 ++++++++++++++++++- internal/service/token_svc.go | 80 +++++++++++++++++------ 11 files changed, 312 insertions(+), 34 deletions(-) diff --git a/internal/handler/auth.go b/internal/handler/auth.go index 9e85fcc..658c7f7 100644 --- a/internal/handler/auth.go +++ b/internal/handler/auth.go @@ -45,7 +45,7 @@ func loginHandler(svc *service.Container) gin.HandlerFunc { "user": resp.User, "tokens": resp.Tokens, }) - svc.Audit.Record(c.Request.Context(), resp.User.ID, "auth.login", resp.User.Username, c.ClientIP(), "") + svc.Audit.RecordBestEffort(resp.User.ID, "auth.login", resp.User.Username, c.ClientIP(), "") } } diff --git a/internal/service/audit.go b/internal/service/audit.go index 80a80f8..cd13bfc 100644 --- a/internal/service/audit.go +++ b/internal/service/audit.go @@ -8,6 +8,7 @@ package service import ( "context" + "time" "go.uber.org/zap" @@ -39,3 +40,16 @@ func (a *AuditService) Record(ctx context.Context, userID, action, target, ip, d a.log.Debug("audit write failed", zap.Error(err)) } } + +// RecordBestEffort writes an audit row off the request path. Login must not be +// held open by SQLite write pressure from scans or background maintenance. +func (a *AuditService) RecordBestEffort(userID, action, target, ip, detail string) { + if a == nil || a.repo == nil || a.repo.Log == nil { + return + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + a.Record(ctx, userID, action, target, ip, detail) + }() +} diff --git a/internal/service/auth.go b/internal/service/auth.go index 1293c34..f9c4435 100644 --- a/internal/service/auth.go +++ b/internal/service/auth.go @@ -163,10 +163,23 @@ func (s *AuthService) Login(ctx context.Context, username, password string) (*Lo if err != nil { return nil, err } - _ = s.repo.User.TouchLogin(ctx, u.ID) + s.touchLoginBestEffort(u.ID) return &LoginResponse{User: u, Tokens: tokens}, nil } +func (s *AuthService) touchLoginBestEffort(userID string) { + if s == nil || s.repo == nil || s.repo.User == nil || strings.TrimSpace(userID) == "" { + return + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := s.repo.User.TouchLogin(ctx, userID); err != nil && s.log != nil { + s.log.Debug("touch login delayed", zap.String("user_id", userID), zap.Error(err)) + } + }() +} + // ChangePassword updates the user password if the old one matches. func (s *AuthService) ChangePassword(ctx context.Context, userID, oldPwd, newPwd string) error { if strings.TrimSpace(newPwd) == "" || len(newPwd) < 6 { diff --git a/internal/service/auth_user_limits_test.go b/internal/service/auth_user_limits_test.go index 9690c39..43f3be8 100644 --- a/internal/service/auth_user_limits_test.go +++ b/internal/service/auth_user_limits_test.go @@ -30,6 +30,11 @@ func newAuthTestServices(t *testing.T) (*repository.Container, *AuthService, *Pr if err := db.AutoMigrate(&model.User{}, &model.UserPermission{}, &model.RefreshToken{}, &model.TelegramBinding{}, &model.Setting{}); err != nil { t.Fatal(err) } + sqlDB, err := db.DB() + if err != nil { + t.Fatal(err) + } + sqlDB.SetMaxOpenConns(1) repos := repository.New(db) cfg := &config.Config{} cfg.Secrets.JWTSecret = "test-secret" diff --git a/internal/service/downloads.go b/internal/service/downloads.go index 0366163..84489b9 100644 --- a/internal/service/downloads.go +++ b/internal/service/downloads.go @@ -17,10 +17,12 @@ package service import ( "context" + "crypto/sha1" "errors" + "fmt" "math" - "os" "net/url" + "os" "path" "path/filepath" "regexp" @@ -821,7 +823,7 @@ func (d *DownloadService) processDownloadSnapshot(ctx context.Context, live []QB // （onTorrentComplete 内部仍受 organize.auto 开关约束，且 // 整理对已存在的目标文件幂等跳过）。 d.prevStates[stateKey] = true - if recentlyCompletedTorrent(torrent, time.Now()) { + if recentlyCompletedTorrent(torrent, time.Now()) && !d.completedTorrentCatchupRecorded(ctx, torrent) { shouldQueue = true } case complete && !wasComplete: @@ -928,6 +930,8 @@ func (d *DownloadService) markCompletedTorrentOrganizeDone(torrent QBitTorrent) // 防止每次启动都把全部历史种子重新过一遍整理流程。 const completedTorrentCatchupWindow = 24 * time.Hour +const completedTorrentCatchupSettingPrefix = "download.auto_organized." + // recentlyCompletedTorrent 报告该种子是否在补整理时间窗内完成。 // qBittorrent 未提供 completion_on 时保守地返回 false。 func recentlyCompletedTorrent(torrent QBitTorrent, now time.Time) bool { @@ -938,6 +942,46 @@ func recentlyCompletedTorrent(torrent QBitTorrent, now time.Time) bool { return now.Sub(completed) <= completedTorrentCatchupWindow } +func (d *DownloadService) completedTorrentCatchupRecorded(ctx context.Context, torrent QBitTorrent) bool { + if d == nil || d.repo == nil || d.repo.Setting == nil { + return false + } + key := completedTorrentCatchupSettingKey(torrent) + if key == "" { + return false + } + value, err := d.repo.Setting.Get(ctx, key) + if err != nil { + return false + } + return parseBoolSetting(value, false) +} + +func (d *DownloadService) markCompletedTorrentCatchupRecorded(ctx context.Context, torrent QBitTorrent) { + if d == nil || d.repo == nil || d.repo.Setting == nil { + return + } + key := completedTorrentCatchupSettingKey(torrent) + if key == "" { + return + } + if err := d.repo.Setting.Set(ctx, key, "true"); err != nil && d.log != nil { + d.log.Debug("mark completed torrent catchup failed", + zap.String("hash", torrent.Hash), + zap.String("name", torrent.Name), + zap.Error(err)) + } +} + +func completedTorrentCatchupSettingKey(torrent QBitTorrent) string { + key := completedTorrentQueueKey(torrent) + if key == "" { + return "" + } + sum := sha1.Sum([]byte(key)) + return completedTorrentCatchupSettingPrefix + fmt.Sprintf("%x", sum[:]) +} + func completedTorrentQueueKey(torrent QBitTorrent) string { hash := strings.ToLower(strings.TrimSpace(torrent.Hash)) if hash != "" { @@ -1054,9 +1098,17 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor zap.Error(err)) return } - if d.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" { + if d.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" && OrganizeResultHasChanges(res) { res.Scans, res.Scrapes = d.scanner.ScanAndScrapeLibrariesForPath(ctx, res.DestPath, "", OrganizeScrapeAfterEnabled(ctx, d.repo)) + } else if d.log != nil && res != nil && !OrganizeResultHasChanges(res) { + d.log.Info("auto organize completed torrent skipped scan; no destination changes", + zap.String("hash", torrent.Hash), + zap.String("source", source), + zap.Int("organized", res.Organized), + zap.Int("replaced", res.Replaced), + zap.Int("skipped", res.Skipped)) } + d.markCompletedTorrentCatchupRecorded(context.Background(), torrent) d.log.Info("auto organize completed torrent finished", zap.String("hash", torrent.Hash), zap.String("source", source), diff --git a/internal/service/downloads_test.go b/internal/service/downloads_test.go index 568fe9c..1d332a4 100644 --- a/internal/service/downloads_test.go +++ b/internal/service/downloads_test.go @@ -206,6 +206,75 @@ func TestDownloadPollCatchesUpRecentlyCompletedTorrents(t *testing.T) { } } +func TestDownloadPollSkipsRecordedCompletedTorrentCatchup(t *testing.T) { + repos := newOrganizerTestRepo(t) + torrent := QBitTorrent{ + Hash: "fresh-complete", + Name: "Fresh Complete S01E01", + Progress: 1, + CompletionOn: time.Now().Add(-time.Hour).Unix(), + } + if err := repos.Setting.Set(t.Context(), completedTorrentCatchupSettingKey(torrent), "true"); err != nil { + t.Fatal(err) + } + svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil) + + svc.processDownloadSnapshot(t.Context(), []QBitTorrent{torrent}, nil) + + if got := len(svc.organizeQueue); got != 0 { + t.Fatalf("recorded completed torrent queued %d organize jobs, want 0", got) + } +} + +func TestAutoOrganizeSkipsScanWhenNoFilesChanged(t *testing.T) { + root := t.TempDir() + src := filepath.Join(root, "downloads", "国产剧", "狂飙.S01E01.2023.1080p.mkv") + dest := filepath.Join(root, "media") + writeOrgFile(t, src, "episode") + + repos := newOrganizerTestRepo(t) + for key, value := range map[string]string{ + "organizer.auto_after_download": "true", + "organize.target_dir": dest, + "organize.transfer_mode": "copy", + } { + if err := repos.Setting.Set(t.Context(), key, value); err != nil { + t.Fatal(err) + } + } + lib := model.Library{Name: "国产剧", Path: filepath.Join(dest, "电视剧", "国产剧"), Type: "tv", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + org := NewOrganizerService(&config.Config{}, zap.NewNop(), repos) + if _, err := org.OrganizeDirectory(t.Context(), OrganizeOptions{ + SourcePath: src, + DestPath: dest, + TransferMode: TransferCopy, + }); err != nil { + t.Fatalf("seed organized destination: %v", err) + } + scanner := NewScannerService(&config.Config{}, zap.NewNop(), repos, NewHub(zap.NewNop()), nil, nil) + svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), org) + svc.SetScanner(scanner) + + svc.onTorrentComplete(t.Context(), QBitTorrent{ + Hash: "done123", + Name: "狂飙.S01E01.2023.1080p", + Progress: 1, + SavePath: filepath.Dir(src), + ContentPath: src, + }) + + var count int64 + if err := repos.DB.Model(&model.Media{}).Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatalf("no-op auto organize triggered scan and created %d media rows, want 0", count) + } +} + func TestCompletedTorrentSourceUsesConfiguredMapping(t *testing.T) { root := t.TempDir() localRoot := filepath.Join(root, "localdl") diff --git a/internal/service/organizer_scan.go b/internal/service/organizer_scan.go index 360029f..17f3bdb 100644 --- a/internal/service/organizer_scan.go +++ b/internal/service/organizer_scan.go @@ -47,6 +47,14 @@ func OrganizeScrapeAfterEnabled(ctx context.Context, repo *repository.Container) return false } +// OrganizeResultHasChanges reports whether an organize run actually changed +// files in the destination library. Skipped duplicates are intentionally not a +// change: scanning after a no-op organize can turn a harmless restart into a +// full library ffprobe sweep. +func OrganizeResultHasChanges(res *OrganizeResult) bool { + return res != nil && (res.Organized > 0 || res.Replaced > 0) +} + // ScanLibrariesForPath recursively scans libraries affected by an organize // destination. If preferredLibraryID is set, only that library is scanned. // Otherwise every enabled library whose path intersects destRoot is scanned; diff --git a/internal/service/path_translation.go b/internal/service/path_translation.go index d2e446a..cd22332 100644 --- a/internal/service/path_translation.go +++ b/internal/service/path_translation.go @@ -18,9 +18,16 @@ func translateClientPath(clientPath string, mappings map[string]string) string { return clean } // 尝试路径映射 + cleanForMatch := filepath.ToSlash(clean) for clientPrefix, localPrefix := range mappings { - if strings.HasPrefix(clean, clientPrefix) { - translated := filepath.Join(localPrefix, strings.TrimPrefix(clean, clientPrefix)) + prefix := strings.TrimRight(filepath.ToSlash(filepath.Clean(clientPrefix)), "/") + if prefix == "" || prefix == "." { + continue + } + if cleanForMatch == prefix || strings.HasPrefix(cleanForMatch, prefix+"/") { + rel := strings.TrimPrefix(cleanForMatch, prefix) + rel = strings.TrimPrefix(rel, "/") + translated := filepath.Join(localPrefix, filepath.FromSlash(rel)) if _, err := os.Stat(translated); err == nil { return translated } diff --git a/internal/service/scheduler.go b/internal/service/scheduler.go index f1e635d..4a90913 100644 --- a/internal/service/scheduler.go +++ b/internal/service/scheduler.go @@ -135,10 +135,10 @@ func (s *SchedulerService) Start(ctx context.Context) { } for _, j := range s.jobs { initialDelay := 15 * time.Second - if j.name == "library_scan" { - // 重启后不立即整库重扫：更新/重启窗口恰是登录高峰，启动 - // 15 秒即全量扫描曾把 CPU/磁盘打满导致无法登录。首轮等满 - // 一个完整周期再跑，平时的每小时节奏不变。 + if j.name == "library_scan" || j.name == "organize_source" { + // 重启后不立即整库重扫/整理下载目录：更新窗口恰是登录高峰， + // 15 秒即全量 walk + ffprobe 曾把 CPU/磁盘打满导致无法登录。 + // 首轮等满一个完整周期再跑，平时节奏不变。 initialDelay = j.interval } go s.loopWithInitialDelay(ctx, j, initialDelay) @@ -483,8 +483,16 @@ func (s *SchedulerService) jobOrganizeSource(ctx context.Context) error { if err != nil { return err } - if s.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" { + if s.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" && OrganizeResultHasChanges(res) { res.Scans, res.Scrapes = s.scanner.ScanAndScrapeLibrariesForPath(ctx, res.DestPath, "", OrganizeScrapeAfterEnabled(ctx, s.repo)) + } else if s.log != nil && res != nil && !OrganizeResultHasChanges(res) { + s.log.Info("scheduled source organize skipped scan; no destination changes", + zap.String("source", res.SourcePath), + zap.String("dest", res.DestPath), + zap.Int("organized", res.Organized), + zap.Int("replaced", res.Replaced), + zap.Int("skipped", res.Skipped), + ) } if s.log != nil && res != nil { s.log.Info("scheduled source organize finished", diff --git a/internal/service/service.go b/internal/service/service.go index 78d1c3e..916967d 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -5,6 +5,7 @@ package service import ( "context" + "strconv" "strings" "time" @@ -268,15 +269,21 @@ func (c *Container) warmMediaSearchIndex(ctx context.Context) { if c == nil || c.Repo == nil || c.Repo.Media == nil { return } + if !mediaSearchWarmupEnabled(ctx, c.Repo) { + if c.Log != nil { + c.Log.Info("media search index warmup disabled") + } + return + } // 错峰：FTS 正常由 media 表触发器实时维护，回填只是升级或异常后的 // 兜底。先让登录、首页等关键路径跑起来，再开始后台补索引。 select { case <-ctx.Done(): return - case <-time.After(30 * time.Second): + case <-time.After(mediaSearchWarmupDelay(ctx, c.Repo)): } - const batchSize = 1000 - const pause = 100 * time.Millisecond + batchSize := mediaSearchWarmupBatchSize(ctx, c.Repo) + pause := mediaSearchWarmupPause(ctx, c.Repo) total := int64(0) for { select { @@ -304,6 +311,59 @@ func (c *Container) warmMediaSearchIndex(ctx context.Context) { } } +func mediaSearchWarmupEnabled(ctx context.Context, repo *repository.Container) bool { + if repo == nil || repo.Setting == nil { + return true + } + value, err := repo.Setting.Get(ctx, "search.index_warmup_enabled") + if err != nil || strings.TrimSpace(value) == "" { + return true + } + return parseBoolSetting(value, true) +} + +func mediaSearchWarmupDelay(ctx context.Context, repo *repository.Container) time.Duration { + seconds := mediaSearchWarmupIntSetting(ctx, repo, "search.index_warmup_delay_seconds", 120) + if seconds < 30 { + seconds = 30 + } + return time.Duration(seconds) * time.Second +} + +func mediaSearchWarmupBatchSize(ctx context.Context, repo *repository.Container) int { + size := mediaSearchWarmupIntSetting(ctx, repo, "search.index_warmup_batch_size", 100) + if size < 10 { + size = 10 + } + if size > 1000 { + size = 1000 + } + return size +} + +func mediaSearchWarmupPause(ctx context.Context, repo *repository.Container) time.Duration { + ms := mediaSearchWarmupIntSetting(ctx, repo, "search.index_warmup_pause_ms", 2000) + if ms < 250 { + ms = 250 + } + return time.Duration(ms) * time.Millisecond +} + +func mediaSearchWarmupIntSetting(ctx context.Context, repo *repository.Container, key string, fallback int) int { + if repo == nil || repo.Setting == nil { + return fallback + } + value, err := repo.Setting.Get(ctx, key) + if err != nil { + return fallback + } + n, err := strconv.Atoi(strings.TrimSpace(value)) + if err != nil || n <= 0 { + return fallback + } + return n +} + func (c *Container) NormalizeCloudLibraryTypes(ctx context.Context) error { if c == nil || c.Repo == nil || c.Repo.Library == nil || c.Repo.DB == nil { return nil diff --git a/internal/service/token_svc.go b/internal/service/token_svc.go index 760842e..fb2f029 100644 --- a/internal/service/token_svc.go +++ b/internal/service/token_svc.go @@ -107,25 +107,17 @@ func (s *TokenService) issuePair(ctx context.Context, userID, role, tier string, TokenHash: tokenHash, ExpiresAt: time.Now().Add(RefreshTokenDuration), } - storeCtx := ctx - cancel := func() {} if bestEffort { - storeCtx, cancel = context.WithTimeout(context.Background(), loginRefreshTokenStoreTimeout) + s.storeRefreshTokenBestEffort(userID, tokenHash, rt.ExpiresAt) + return &TokenPair{ + AccessToken: accessToken, + RefreshToken: refreshToken, + ExpiresIn: int64(AccessTokenDuration.Seconds()), + TokenType: "Bearer", + }, nil } - err = s.storeRefreshToken(storeCtx, rt) - cancel() - if err != nil { - if !bestEffort { - return nil, err - } - if s.log != nil { - s.log.Warn("refresh token store delayed; login will continue", - zap.String("user_id", userID), - zap.Error(err)) - } - if s.trackDelayedStore(userID, tokenHash, rt.ExpiresAt) { - go s.storeRefreshTokenEventually(userID, tokenHash, rt.ExpiresAt) - } + if err := s.storeRefreshToken(ctx, rt); err != nil { + return nil, err } return &TokenPair{ @@ -136,6 +128,56 @@ func (s *TokenService) issuePair(ctx context.Context, userID, role, tier string, }, nil } +func (s *TokenService) storeRefreshTokenBestEffort(userID, tokenHash string, expiresAt time.Time) { + if !s.trackDelayedStore(userID, tokenHash, expiresAt) { + return + } + done := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + done <- s.storeRefreshToken(ctx, &model.RefreshToken{ + UserID: userID, + TokenHash: tokenHash, + ExpiresAt: expiresAt, + }) + }() + select { + case err := <-done: + s.finishBestEffortRefreshTokenStore(userID, tokenHash, expiresAt, err) + case <-time.After(loginRefreshTokenStoreTimeout): + if s.log != nil { + s.log.Warn("refresh token store delayed; login will continue", + zap.String("user_id", userID), + zap.Error(context.DeadlineExceeded)) + } + go func() { + err := <-done + s.finishBestEffortRefreshTokenStore(userID, tokenHash, expiresAt, err) + }() + } +} + +func (s *TokenService) finishBestEffortRefreshTokenStore(userID, tokenHash string, expiresAt time.Time, err error) { + if err == nil { + s.untrackDelayedStore(userID, tokenHash) + return + } + if repository.IsSQLiteBusyError(err) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) { + if s.log != nil { + s.log.Warn("refresh token store delayed; login will continue", + zap.String("user_id", userID), + zap.Error(err)) + } + s.storeRefreshTokenEventually(userID, tokenHash, expiresAt) + return + } + s.untrackDelayedStore(userID, tokenHash) + if s.log != nil { + s.log.Warn("refresh token delayed store failed permanently", zap.String("user_id", userID), zap.Error(err)) + } +} + func (s *TokenService) storeRefreshToken(ctx context.Context, rt *model.RefreshToken) error { if err := s.repo.RefreshToken.Create(ctx, rt); err != nil { return err @@ -148,7 +190,7 @@ func (s *TokenService) storeRefreshToken(ctx context.Context, rt *model.RefreshT func (s *TokenService) storeRefreshTokenEventually(userID, tokenHash string, expiresAt time.Time) { defer s.untrackDelayedStore(userID, tokenHash) - delay := 5 * time.Second + delay := time.Second for attempt := 1; attempt <= 8; attempt++ { timer := time.NewTimer(delay) <-timer.C @@ -313,7 +355,7 @@ func (s *TokenService) Refresh(ctx context.Context, refreshToken string) (*Token s.untrackDelayedStore(rt.UserID, tokenHash) // 签发新的令牌对 - return s.IssuePair(ctx, user.ID, user.Role, user.Tier) + return s.IssuePairBestEffort(ctx, user.ID, user.Role, user.Tier) } // RevokeAll 撤销用户的所有 Refresh Token（用于登出）。