Merge fix/resource-login-organize-playback

internal/database/database.go

@@ -3,9 +3,9 @@
 package database
 
 import (
+	"context"
 	"fmt"
 	"path/filepath"
-	"sync"
 
 	"github.com/glebarez/sqlite"
 	"go.uber.org/zap"
@@ -57,12 +57,23 @@ func installSQLiteWriteGate(db *gorm.DB) {
 	if db == nil {
 		return
 	}
-	gate := &sqliteWriteGate{}
+	const lockedKey = "mediastation:sqlite_write_locked"
+	gate := newSQLiteWriteGate()
 	lock := func(tx *gorm.DB) {
-		gate.Lock()
+		ctx := context.Background()
+		if tx.Statement != nil && tx.Statement.Context != nil {
+			ctx = tx.Statement.Context
+		}
+		if err := gate.Lock(ctx); err != nil {
+			_ = tx.AddError(err)
+			return
+		}
+		tx.InstanceSet(lockedKey, struct{}{})
 	}
 	unlock := func(tx *gorm.DB) {
-		gate.Unlock()
+		if _, ok := tx.InstanceGet(lockedKey); ok {
+			gate.Unlock()
+		}
 	}
 	_ = db.Callback().Create().Before("gorm:create").Register("mediastation:sqlite_write_lock", lock)
 	_ = db.Callback().Create().After("gorm:create").Register("mediastation:sqlite_write_unlock", unlock)
@@ -74,16 +85,41 @@ func installSQLiteWriteGate(db *gorm.DB) {
 	_ = db.Callback().Raw().After("gorm:raw").Register("mediastation:sqlite_write_unlock", unlock)
 }
 
+// sqliteWriteGate 串行化进程内的 SQLite 写操作，避免多连接写竞争触发
+// SQLITE_BUSY。Lock 尊重语句自身的 context：此前用 sync.Mutex 时，一条
+// 长写语句（如 FTS 回填批次）会让登录等关键写操作无限期排队——客户端
+// 早已超时断开，goroutine 还挂在互斥锁上。现在等待方可随 context 取消
+// 及时失败，不再把整个进程的写路径拖死。
 type sqliteWriteGate struct {
-	mu sync.Mutex
+	ch chan struct{}
 }
 
-func (g *sqliteWriteGate) Lock() {
-	g.mu.Lock()
+func newSQLiteWriteGate() *sqliteWriteGate {
+	return &sqliteWriteGate{ch: make(chan struct{}, 1)}
+}
+
+func (g *sqliteWriteGate) Lock(ctx context.Context) error {
+	select {
+	case g.ch <- struct{}{}:
+		return nil
+	default:
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	select {
+	case g.ch <- struct{}{}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
 func (g *sqliteWriteGate) Unlock() {
-	g.mu.Unlock()
+	select {
+	case <-g.ch:
+	default:
+	}
 }
 
 func buildDSN(cfg *config.Config) string {
@@ -139,9 +175,31 @@ func ensurePerformanceIndexes(db *gorm.DB) error {
 	return nil
 }
 
+// mediaSearchIndexSchemaVersion 标识 FTS 索引的物理布局版本。
+// v2：FTS 行的 rowid 与 media.rowid 对齐，并由触发器实时维护。
+const mediaSearchIndexSchemaVersion = 2
+
 func ensureMediaSearchIndex(db *gorm.DB) error {
-	if mediaSearchIndexNeedsRebuild(db) {
-		_ = db.Exec(`DROP TABLE IF EXISTS media_search_fts`).Error
+	if err := db.Exec(`CREATE TABLE IF NOT EXISTS media_search_meta (id INTEGER PRIMARY KEY CHECK (id = 1), version INTEGER NOT NULL)`).Error; err != nil {
+		return nil
+	}
+	var version int
+	_ = db.Raw(`SELECT version FROM media_search_meta WHERE id = 1`).Scan(&version).Error
+	if version != mediaSearchIndexSchemaVersion {
+		// 旧版（v1）FTS 表按 UNINDEXED 的 media_id 寻址。FTS5 的普通列
+		// 不支持索引查找，按 media_id 的 DELETE / NOT EXISTS 都是整表
+		// 扫描：十几万行的库每次启动回填要做上百亿次行访问，纯 Go
+		// sqlite 直接把 CPU 钉满数小时，并隔着全局写锁拖死登录。
+		// v2 起 FTS 行的 rowid 与 media.rowid 对齐，所有寻址走 rowid
+		// 点查，索引一致性交给下方触发器维护。
+		for _, stmt := range []string{
+			`DROP TRIGGER IF EXISTS media_search_fts_ai`,
+			`DROP TRIGGER IF EXISTS media_search_fts_au`,
+			`DROP TRIGGER IF EXISTS media_search_fts_ad`,
+			`DROP TABLE IF EXISTS media_search_fts`,
+		} {
+			_ = db.Exec(stmt).Error
+		}
 	}
 	if err := db.Exec(`CREATE VIRTUAL TABLE IF NOT EXISTS media_search_fts USING fts5(media_id UNINDEXED, title, original_name, path, genres, tokenize='trigram')`).Error; err != nil {
 		if fallbackErr := db.Exec(`CREATE VIRTUAL TABLE IF NOT EXISTS media_search_fts USING fts5(media_id UNINDEXED, title, original_name, path, genres, tokenize='unicode61')`).Error; fallbackErr != nil {
@@ -151,22 +209,35 @@ func ensureMediaSearchIndex(db *gorm.DB) error {
 			return nil
 		}
 	}
-	return nil
-}
-
-func mediaSearchIndexNeedsRebuild(db *gorm.DB) bool {
-	var cols []struct {
-		Name string
-	}
-	if err := db.Raw(`PRAGMA table_info(media_search_fts)`).Scan(&cols).Error; err != nil || len(cols) == 0 {
-		return false
-	}
-	for _, col := range cols {
-		if col.Name == "genres" {
-			return false
+	// 触发器让 FTS 与 media 行保持同步（新增/标题刮削改写/软删/恢复/
+	// 硬删全覆盖），应用层不再需要按 media_id 手工刷新索引——也顺带
+	// 修复了刮削直写 Updates() 后新标题搜不到的问题。
+	for _, stmt := range []string{
+		`CREATE TRIGGER IF NOT EXISTS media_search_fts_ai AFTER INSERT ON media WHEN new.deleted_at IS NULL BEGIN
+			DELETE FROM media_search_fts WHERE rowid = new.rowid;
+			INSERT INTO media_search_fts(rowid, media_id, title, original_name, path, genres)
+			VALUES (new.rowid, new.id, COALESCE(new.title, ''), COALESCE(new.original_name, ''), COALESCE(new.path, ''), COALESCE(new.genres, ''));
+		END`,
+		`CREATE TRIGGER IF NOT EXISTS media_search_fts_au AFTER UPDATE OF title, original_name, path, genres, deleted_at ON media BEGIN
+			DELETE FROM media_search_fts WHERE rowid = old.rowid;
+			INSERT INTO media_search_fts(rowid, media_id, title, original_name, path, genres)
+			SELECT new.rowid, new.id, COALESCE(new.title, ''), COALESCE(new.original_name, ''), COALESCE(new.path, ''), COALESCE(new.genres, '')
+			WHERE new.deleted_at IS NULL;
+		END`,
+		`CREATE TRIGGER IF NOT EXISTS media_search_fts_ad AFTER DELETE ON media BEGIN
+			DELETE FROM media_search_fts WHERE rowid = old.rowid;
+		END`,
+	} {
+		if err := db.Exec(stmt).Error; err != nil {
+			return err
 		}
 	}
-	return true
+	if version != mediaSearchIndexSchemaVersion {
+		if err := db.Exec(`INSERT INTO media_search_meta(id, version) VALUES (1, ?) ON CONFLICT(id) DO UPDATE SET version = excluded.version`, mediaSearchIndexSchemaVersion).Error; err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func enforceTelegramBindingOneToOne(db *gorm.DB) error {

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(), "")
 	}
 }
 

internal/repository/repository.go

@@ -323,7 +323,6 @@ func (r *MediaRepository) upsert(ctx context.Context, m *model.Media) error {
 			m.ScrapeStatus = "pending"
 		}
 		if createErr := r.db.WithContext(ctx).Create(m).Error; createErr == nil {
-			_ = r.refreshSearchIndex(ctx, m.ID)
 			return nil
 		} else if retryErr := r.db.WithContext(ctx).Unscoped().Where("path = ?", m.Path).First(&existing).Error; retryErr != nil {
 			return createErr
@@ -426,7 +425,6 @@ func (r *MediaRepository) upsert(ctx context.Context, m *model.Media) error {
 		Where("id = ?", existing.ID).Updates(updates).Error; err != nil {
 		return err
 	}
-	_ = r.refreshSearchIndex(ctx, existing.ID)
 	// 回写 ID / 不可变字段，让 caller 拿到完整的现有行。
 	*m = existing
 	return nil
@@ -518,7 +516,7 @@ func (r *MediaRepository) searchFilteredFTS(ctx context.Context, query string, o
 	var items []model.Media
 	q := r.db.WithContext(ctx).
 		Table("media").
-		Joins("JOIN media_search_fts ON media_search_fts.media_id = media.id").
+		Joins("JOIN media_search_fts ON media_search_fts.rowid = media.rowid").
 		Where("media.deleted_at IS NULL").
 		Where("media_search_fts MATCH ?", ftsQuery)
 	q = applyQualifiedMediaQueryFilter(q, filter)
@@ -625,23 +623,6 @@ func escapeLike(value string) string {
 	return value
 }
 
-func (r *MediaRepository) refreshSearchIndex(ctx context.Context, mediaID string) error {
-	if strings.TrimSpace(mediaID) == "" {
-		return nil
-	}
-	if !r.searchIndexEnabled(ctx) {
-		return nil
-	}
-	tx := r.db.WithContext(ctx)
-	_ = tx.Exec(`DELETE FROM media_search_fts WHERE media_id = ?`, mediaID).Error
-	return tx.Exec(`
-INSERT INTO media_search_fts(media_id, title, original_name, path, genres)
-SELECT id, COALESCE(title, ''), COALESCE(original_name, ''), COALESCE(path, ''), COALESCE(genres, '')
-FROM media
-WHERE id = ? AND deleted_at IS NULL
-`, mediaID).Error
-}
-
 func (r *MediaRepository) BackfillSearchIndex(ctx context.Context, batchLimit int) (int64, error) {
 	if batchLimit <= 0 {
 		batchLimit = 1000
@@ -649,15 +630,19 @@ func (r *MediaRepository) BackfillSearchIndex(ctx context.Context, batchLimit in
 	if !r.searchIndexEnabled(ctx) {
 		return 0, nil
 	}
+	// 关键性能点：FTS5 普通列（含 UNINDEXED）不支持索引查找，按
+	// media_id 做 NOT EXISTS 是对 FTS 表的整表扫描，再叠加 ORDER BY
+	// 后每个批次都要对全部 media 行探测一遍——大库一次启动回填等于
+	// 上百亿次行访问，曾把 CPU 钉满数小时。v2 布局下 FTS 行 rowid 与
+	// media.rowid 对齐，NOT EXISTS 走 rowid 点查，且无需排序。
 	res := r.db.WithContext(ctx).Exec(`
-INSERT INTO media_search_fts(media_id, title, original_name, path, genres)
-SELECT m.id, COALESCE(m.title, ''), COALESCE(m.original_name, ''), COALESCE(m.path, ''), COALESCE(m.genres, '')
+INSERT INTO media_search_fts(rowid, media_id, title, original_name, path, genres)
+SELECT m.rowid, m.id, COALESCE(m.title, ''), COALESCE(m.original_name, ''), COALESCE(m.path, ''), COALESCE(m.genres, '')
 FROM media AS m
 WHERE m.deleted_at IS NULL
   AND NOT EXISTS (
-    SELECT 1 FROM media_search_fts AS f WHERE f.media_id = m.id
+    SELECT 1 FROM media_search_fts AS f WHERE f.rowid = m.rowid
   )
-ORDER BY m.created_at DESC
 LIMIT ?
 `, batchLimit)
 	return res.RowsAffected, res.Error
@@ -679,18 +664,13 @@ func (r *MediaRepository) searchIndexEnabled(ctx context.Context) bool {
 
 // DeleteByLibrary purges all media tied to a library.
 func (r *MediaRepository) DeleteByLibrary(ctx context.Context, libraryID string) error {
-	if r.searchIndexEnabled(ctx) {
-		_ = r.db.WithContext(ctx).Exec(`DELETE FROM media_search_fts WHERE media_id IN (SELECT id FROM media WHERE library_id = ?)`, libraryID).Error
-	}
+	// FTS 行由 media 表上的触发器同步清理（软删/硬删都覆盖）。
 	return r.db.WithContext(ctx).Where("library_id = ?", libraryID).Delete(&model.Media{}).Error
 }
 
 // PurgeByLibrary permanently removes media tied to a library. Used for virtual
 // cloud mounts where "remove mount" must not populate the recycle bin.
 func (r *MediaRepository) PurgeByLibrary(ctx context.Context, libraryID string) error {
-	if r.searchIndexEnabled(ctx) {
-		_ = r.db.WithContext(ctx).Exec(`DELETE FROM media_search_fts WHERE media_id IN (SELECT id FROM media WHERE library_id = ?)`, libraryID).Error
-	}
 	return r.db.WithContext(ctx).Unscoped().Where("library_id = ?", libraryID).Delete(&model.Media{}).Error
 }
 

internal/repository/repository_test.go

@@ -153,12 +153,17 @@ func TestMediaSearchIndexBackfillRunsInBatches(t *testing.T) {
 	}).Error; err != nil {
 		t.Fatal(err)
 	}
-	var before int64
-	if err := repos.DB.Raw(`SELECT COUNT(*) FROM media_search_fts`).Scan(&before).Error; err != nil {
+	// 插入触发器应当同步维护 FTS 行。
+	var indexed int64
+	if err := repos.DB.Raw(`SELECT COUNT(*) FROM media_search_fts`).Scan(&indexed).Error; err != nil {
 		t.Fatal(err)
 	}
-	if before != 0 {
-		t.Fatalf("startup migrate should not synchronously backfill FTS, got %d rows", before)
+	if indexed != 1 {
+		t.Fatalf("insert trigger should index new media, got %d rows", indexed)
+	}
+	// 清空 FTS 模拟旧库升级后索引缺失，回填应按批补齐且 rowid 对齐。
+	if err := repos.DB.Exec(`DELETE FROM media_search_fts`).Error; err != nil {
+		t.Fatal(err)
 	}
 	n, err := repos.Media.BackfillSearchIndex(t.Context(), 1)
 	if err != nil {
@@ -167,11 +172,22 @@ func TestMediaSearchIndexBackfillRunsInBatches(t *testing.T) {
 	if n != 1 {
 		t.Fatalf("backfilled rows = %d, want 1", n)
 	}
+	var aligned int64
+	if err := repos.DB.Raw(`SELECT COUNT(*) FROM media_search_fts f JOIN media m ON f.rowid = m.rowid AND f.media_id = m.id`).Scan(&aligned).Error; err != nil {
+		t.Fatal(err)
+	}
+	if aligned != 1 {
+		t.Fatalf("fts rows aligned with media rowid = %d, want 1", aligned)
+	}
+	// 软删除后触发器应清理对应 FTS 行，避免搜索命中已删媒体。
+	if err := repos.DB.Delete(&model.Media{}, "id = ?", "m-backfill").Error; err != nil {
+		t.Fatal(err)
+	}
 	var after int64
 	if err := repos.DB.Raw(`SELECT COUNT(*) FROM media_search_fts`).Scan(&after).Error; err != nil {
 		t.Fatal(err)
 	}
-	if after != 1 {
-		t.Fatalf("fts rows = %d, want 1", after)
+	if after != 0 {
+		t.Fatalf("soft delete should drop fts row, got %d", after)
 	}
 }

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)
+	}()
+}

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 {

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"

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),

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")

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;

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
 			}

internal/service/scanner.go

@@ -1491,13 +1491,20 @@ func (s *ScannerService) mediaPathExists(ctx context.Context, path string) bool
 }
 
 func (s *ScannerService) pruneMissingMedia(ctx context.Context, libraryID string, seen map[string]struct{}) (int64, error) {
-	var rows []model.Media
+	// 只取 id/path，并把删除按批提交：此前整表载入完整 Media 结构体、
+	// 每行一条 DELETE，大库 prune 既费内存又长期占用写锁。
+	var rows []struct {
+		ID   string
+		Path string
+	}
 	if err := s.repo.DB.WithContext(ctx).
+		Model(&model.Media{}).
+		Select("id, path").
 		Where("library_id = ?", libraryID).
 		Find(&rows).Error; err != nil {
 		return 0, err
 	}
-	var removed int64
+	stale := make([]string, 0)
 	for _, row := range rows {
 		if row.Path == "" {
 			continue
@@ -1510,9 +1517,26 @@ func (s *ScannerService) pruneMissingMedia(ctx context.Context, libraryID string
 		} else if !os.IsNotExist(err) {
 			continue
 		}
-		res := s.repo.DB.WithContext(ctx).
-			Where("id = ?", row.ID).
-			Delete(&model.Media{})
+		stale = append(stale, row.ID)
+	}
+	return s.deleteMediaByIDs(ctx, stale, false)
+}
+
+// deleteMediaByIDs removes media rows in fixed-size batches so each write
+// transaction stays short and the global write gate is released frequently.
+func (s *ScannerService) deleteMediaByIDs(ctx context.Context, ids []string, hard bool) (int64, error) {
+	const batch = 500
+	var removed int64
+	for i := 0; i < len(ids); i += batch {
+		end := i + batch
+		if end > len(ids) {
+			end = len(ids)
+		}
+		q := s.repo.DB.WithContext(ctx)
+		if hard {
+			q = q.Unscoped()
+		}
+		res := q.Where("id IN ?", ids[i:end]).Delete(&model.Media{})
 		if res.Error != nil {
 			return removed, res.Error
 		}
@@ -1522,27 +1546,25 @@ func (s *ScannerService) pruneMissingMedia(ctx context.Context, libraryID string
 }
 
 func (s *ScannerService) pruneMissingCloudMedia(ctx context.Context, libraryID string, seen map[string]struct{}) (int64, error) {
-	var rows []model.Media
+	var rows []struct {
+		ID   string
+		Path string
+	}
 	if err := s.repo.DB.WithContext(ctx).
+		Model(&model.Media{}).
+		Select("id, path").
 		Where("library_id = ? AND path LIKE ?", libraryID, "cloud://%").
 		Find(&rows).Error; err != nil {
 		return 0, err
 	}
-	var removed int64
+	stale := make([]string, 0)
 	for _, row := range rows {
 		if _, ok := seen[row.Path]; ok {
 			continue
 		}
-		res := s.repo.DB.WithContext(ctx).
-			Unscoped().
-			Where("id = ?", row.ID).
-			Delete(&model.Media{})
-		if res.Error != nil {
-			return removed, res.Error
-		}
-		removed += res.RowsAffected
+		stale = append(stale, row.ID)
 	}
-	return removed, nil
+	return s.deleteMediaByIDs(ctx, stale, true)
 }
 
 func parseCloudLibraryPath(raw string) (typ, dirID string, ok bool) {

internal/service/scheduler.go

@@ -134,7 +134,14 @@ func (s *SchedulerService) Start(ctx context.Context) {
 		},
 	}
 	for _, j := range s.jobs {
-		go s.loop(ctx, j)
+		initialDelay := 15 * time.Second
+		if j.name == "library_scan" || j.name == "organize_source" {
+			// 重启后不立即整库重扫/整理下载目录：更新窗口恰是登录高峰，
+			// 15 秒即全量 walk + ffprobe 曾把 CPU/磁盘打满导致无法登录。
+			// 首轮等满一个完整周期再跑，平时节奏不变。
+			initialDelay = j.interval
+		}
+		go s.loopWithInitialDelay(ctx, j, initialDelay)
 	}
 }
 
@@ -259,6 +266,13 @@ func (s *SchedulerService) jobScanLibraries(ctx context.Context) error {
 		if !l.Enabled {
 			continue
 		}
+		if _, ok := ParseCloudLibraryMount(l.Path); ok {
+			// 云盘库由 cloud_sync 任务在夜间窗口低频同步；周期性整库
+			// 重扫只面向本地磁盘库。否则十几个云盘库每小时全量遍历
+			// 会把 CPU/网络长期吃满，还会占住唯一的云扫描槽位，让
+			// 手动扫描看起来一直"卡死"在排队。
+			continue
+		}
 		if _, err := s.scanner.ScanLibrary(ctx, l.ID); err != nil {
 			s.log.Warn("scheduled scan failed",
 				zap.String("library", l.ID), zap.Error(err))
@@ -469,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",

internal/service/service.go

@@ -5,6 +5,7 @@ package service
 
 import (
 	"context"
+	"strconv"
 	"strings"
 	"time"
 
@@ -268,8 +269,21 @@ func (c *Container) warmMediaSearchIndex(ctx context.Context) {
 	if c == nil || c.Repo == nil || c.Repo.Media == nil {
 		return
 	}
-	const batchSize = 1000
-	const pause = 100 * time.Millisecond
+	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(mediaSearchWarmupDelay(ctx, c.Repo)):
+	}
+	batchSize := mediaSearchWarmupBatchSize(ctx, c.Repo)
+	pause := mediaSearchWarmupPause(ctx, c.Repo)
 	total := int64(0)
 	for {
 		select {
@@ -297,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

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（用于登出）。