diff --git a/config.example.yaml b/config.example.yaml index 156201a..3cd0b03 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -29,8 +29,8 @@ database: wal_mode: true busy_timeout: 5000 cache_size: -20000 # KB (negative) - max_open_conns: 1 - max_idle_conns: 1 + max_open_conns: 4 # small WAL pool; keeps login/playback responsive during scans + max_idle_conns: 2 secrets: # When empty, a 32-byte random secret is generated on first start and diff --git a/internal/config/config.go b/internal/config/config.go index 03f3326..9e3e694 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -21,6 +21,11 @@ import ( // EnvPrefix 是所有环境变量驱动的覆盖使用的前缀。 const EnvPrefix = "MEDIASTATION" +const ( + defaultDatabaseMaxOpenConns = 4 + defaultDatabaseMaxIdleConns = 2 +) + // Config 是根配置聚合。 type Config struct { App AppConfig `mapstructure:"app"` @@ -226,8 +231,8 @@ func setDefaults(v *viper.Viper) { v.SetDefault("database.wal_mode", true) v.SetDefault("database.busy_timeout", 5000) v.SetDefault("database.cache_size", -20000) - v.SetDefault("database.max_open_conns", 1) - v.SetDefault("database.max_idle_conns", 1) + v.SetDefault("database.max_open_conns", defaultDatabaseMaxOpenConns) + v.SetDefault("database.max_idle_conns", defaultDatabaseMaxIdleConns) v.SetDefault("secrets.jwt_secret", "") @@ -303,6 +308,15 @@ func (c *Config) normalize() error { if c.Database.DBPath == "" { c.Database.DBPath = filepath.Join(c.App.DataDir, "mediastation.db") } + if c.Database.MaxOpenConns <= 1 { + c.Database.MaxOpenConns = defaultDatabaseMaxOpenConns + } + if c.Database.MaxIdleConns <= 0 || c.Database.MaxIdleConns > c.Database.MaxOpenConns { + c.Database.MaxIdleConns = defaultDatabaseMaxIdleConns + if c.Database.MaxIdleConns > c.Database.MaxOpenConns { + c.Database.MaxIdleConns = c.Database.MaxOpenConns + } + } if c.Cache.CacheDir == "" { c.Cache.CacheDir = filepath.Join(c.App.DataDir, "cache") } @@ -335,7 +349,3 @@ func asConfigFileNotFound(err error, target *viper.ConfigFileNotFoundError) bool } return false } - - - - diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 0a53364..182afd8 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -28,6 +28,12 @@ func TestLoadDefaults(t *testing.T) { if cfg.Database.DBPath == "" { t.Fatalf("expected non-empty DBPath") } + if cfg.Database.MaxOpenConns != defaultDatabaseMaxOpenConns { + t.Fatalf("expected default MaxOpenConns %d, got %d", defaultDatabaseMaxOpenConns, cfg.Database.MaxOpenConns) + } + if cfg.Database.MaxIdleConns != defaultDatabaseMaxIdleConns { + t.Fatalf("expected default MaxIdleConns %d, got %d", defaultDatabaseMaxIdleConns, cfg.Database.MaxIdleConns) + } if cfg.Secrets.JWTSecret == "" { t.Fatalf("expected auto-generated JWT secret") } @@ -62,3 +68,23 @@ func TestEnvOverride(t *testing.T) { t.Fatalf("expected port 9090 from env, got %d", cfg.App.Port) } } + +func TestLoadHealsHistoricalSingleConnectionDatabaseConfig(t *testing.T) { + dir := t.TempDir() + wd, _ := os.Getwd() + defer func() { _ = os.Chdir(wd) }() + if err := os.Chdir(dir); err != nil { + t.Fatalf("chdir: %v", err) + } + if err := os.WriteFile("config.yaml", []byte("database:\n max_open_conns: 1\n max_idle_conns: 1\n"), 0o644); err != nil { + t.Fatal(err) + } + + cfg, err := Load() + if err != nil { + t.Fatalf("Load() error: %v", err) + } + if cfg.Database.MaxOpenConns != defaultDatabaseMaxOpenConns { + t.Fatalf("expected historical MaxOpenConns=1 to heal to %d, got %d", defaultDatabaseMaxOpenConns, cfg.Database.MaxOpenConns) + } +} diff --git a/internal/database/database.go b/internal/database/database.go index 96f69e1..90cda72 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -86,6 +86,8 @@ func AutoMigrate(db *gorm.DB) error { func ensurePerformanceIndexes(db *gorm.DB) error { statements := []string{ + `CREATE INDEX IF NOT EXISTS idx_media_created_active ON media(created_at DESC) WHERE deleted_at IS NULL`, + `CREATE INDEX IF NOT EXISTS idx_media_episode_created_active ON media(created_at DESC) WHERE deleted_at IS NULL AND (season_num > 0 OR episode_num > 0)`, `CREATE INDEX IF NOT EXISTS idx_media_library_created_active ON media(library_id, created_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_media_library_episode_active ON media(library_id, season_num, episode_num, created_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_media_series_active ON media(series_id, season_num, episode_num) WHERE deleted_at IS NULL`, diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 96e380b..5c4bdef 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -58,6 +58,8 @@ func TestEnsurePerformanceIndexesCreatesHotPathIndexes(t *testing.T) { t.Fatal(err) } for _, name := range []string{ + "idx_media_created_active", + "idx_media_episode_created_active", "idx_media_library_created_active", "idx_media_library_episode_active", "idx_favorites_user_media_active", diff --git a/internal/service/downloads.go b/internal/service/downloads.go index ecfbfd4..4f8bdb2 100644 --- a/internal/service/downloads.go +++ b/internal/service/downloads.go @@ -44,10 +44,14 @@ type DownloadService struct { scanner *ScannerService site *SiteService - mu sync.Mutex - stopCh chan struct{} - pollOnce sync.Once - prevStates map[string]bool // hash -> wasCompleted + mu sync.Mutex + stopCh chan struct{} + pollOnce sync.Once + organizeOnce sync.Once + prevStates map[string]bool // hash -> wasCompleted + pollInitialized bool + organizeQueue chan QBitTorrent + organizeQueued map[string]struct{} } func (d *DownloadService) SetScanner(scanner *ScannerService) { @@ -58,6 +62,10 @@ var torrentEpisodeToken = regexp.MustCompile(`(?i)e\d{1,3}`) const settingDownloadClientsManaged = "download_clients.managed" +const completedTorrentOrganizeQueueSize = 64 + +var completedTorrentOrganizeCooldown = 3 * time.Second + // ErrDownloadAlreadyExists tells callers that the requested resource is already // tracked locally or present in qBittorrent. Subscriptions treat this as a // successful dedup hit, not as a retryable enqueue failure. @@ -132,14 +140,16 @@ func NewDownloadService(log *zap.Logger, repo *repository.Container, hub *Hub, o siteSvc = site[0] } return &DownloadService{ - log: log, - repo: repo, - hub: hub, - qb: NewQBitClient(log, QBitConfig{}), - organizer: organizer, - site: siteSvc, - prevStates: make(map[string]bool), - stopCh: make(chan struct{}), + log: log, + repo: repo, + hub: hub, + qb: NewQBitClient(log, QBitConfig{}), + organizer: organizer, + site: siteSvc, + prevStates: make(map[string]bool), + organizeQueue: make(chan QBitTorrent, completedTorrentOrganizeQueueSize), + organizeQueued: make(map[string]struct{}), + stopCh: make(chan struct{}), } } @@ -147,6 +157,7 @@ func NewDownloadService(log *zap.Logger, repo *repository.Container, hub *Hub, o func (d *DownloadService) Start(ctx context.Context) { d.pollOnce.Do(func() { _ = d.ReloadConfig(ctx) + d.startAutoOrganizeWorker(ctx) go d.poll(ctx) }) } @@ -709,7 +720,11 @@ func (d *DownloadService) Delete(ctx context.Context, hash string, withFiles boo return err } d.markDownloadTaskDeleted(ctx, torrentName) - delete(d.prevStates, hash) + stateKey := strings.ToLower(hash) + d.mu.Lock() + delete(d.prevStates, stateKey) + delete(d.organizeQueued, stateKey) + d.mu.Unlock() return nil } @@ -770,21 +785,152 @@ func (d *DownloadService) poll(ctx context.Context) { } rows, _ := d.repo.Download.List(ctx) taskByKey := tasksByIdentity(rows) - // Detect completed downloads and trigger organize - for _, t := range live { - hash := t.Hash - complete := t.Progress >= 1.0 - d.syncDownloadTaskProgress(ctx, t, taskByKey) - if complete && !d.prevStates[hash] { - // Just completed: trigger organize - go d.onTorrentComplete(ctx, t) - } - d.prevStates[hash] = complete - } + d.processDownloadSnapshot(ctx, live, taskByKey) d.hub.Publish("download", map[string]any{"torrents": live}) } } +func (d *DownloadService) processDownloadSnapshot(ctx context.Context, live []QBitTorrent, taskByKey map[string]model.DownloadTask) { + d.mu.Lock() + if d.prevStates == nil { + d.prevStates = make(map[string]bool) + } + firstSnapshot := !d.pollInitialized + if firstSnapshot { + d.pollInitialized = true + } + d.mu.Unlock() + + for _, torrent := range live { + stateKey := completedTorrentQueueKey(torrent) + complete := torrent.Progress >= 1.0 + d.syncDownloadTaskProgress(ctx, torrent, taskByKey) + if stateKey == "" { + continue + } + + shouldQueue := false + d.mu.Lock() + wasComplete, wasSeen := d.prevStates[stateKey] + switch { + case complete && (firstSnapshot || !wasSeen): + d.prevStates[stateKey] = true + case complete && !wasComplete: + shouldQueue = true + case complete: + d.prevStates[stateKey] = true + default: + d.prevStates[stateKey] = false + } + d.mu.Unlock() + + if shouldQueue && d.enqueueCompletedTorrent(torrent) { + d.mu.Lock() + d.prevStates[stateKey] = true + d.mu.Unlock() + } + } +} + +func (d *DownloadService) startAutoOrganizeWorker(ctx context.Context) { + d.mu.Lock() + if d.organizeQueue == nil { + d.organizeQueue = make(chan QBitTorrent, completedTorrentOrganizeQueueSize) + } + if d.organizeQueued == nil { + d.organizeQueued = make(map[string]struct{}) + } + d.mu.Unlock() + d.organizeOnce.Do(func() { + go d.autoOrganizeWorker(ctx) + }) +} + +func (d *DownloadService) enqueueCompletedTorrent(torrent QBitTorrent) bool { + key := completedTorrentQueueKey(torrent) + if key == "" { + return false + } + d.mu.Lock() + if d.organizeQueue == nil { + d.organizeQueue = make(chan QBitTorrent, completedTorrentOrganizeQueueSize) + } + if d.organizeQueued == nil { + d.organizeQueued = make(map[string]struct{}) + } + if _, ok := d.organizeQueued[key]; ok { + d.mu.Unlock() + return true + } + select { + case d.organizeQueue <- torrent: + d.organizeQueued[key] = struct{}{} + d.mu.Unlock() + return true + default: + d.mu.Unlock() + if d.log != nil { + d.log.Warn("auto organize queue full; will retry completed torrent later", + zap.String("hash", torrent.Hash), + zap.String("name", torrent.Name)) + } + return false + } +} + +func (d *DownloadService) autoOrganizeWorker(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case <-d.stopCh: + return + case torrent := <-d.organizeQueue: + d.onTorrentComplete(ctx, torrent) + d.markCompletedTorrentOrganizeDone(torrent) + if completedTorrentOrganizeCooldown <= 0 { + continue + } + timer := time.NewTimer(completedTorrentOrganizeCooldown) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-d.stopCh: + timer.Stop() + return + case <-timer.C: + } + } + } +} + +func (d *DownloadService) markCompletedTorrentOrganizeDone(torrent QBitTorrent) { + key := completedTorrentQueueKey(torrent) + if key == "" { + return + } + d.mu.Lock() + delete(d.organizeQueued, key) + d.mu.Unlock() +} + +func completedTorrentQueueKey(torrent QBitTorrent) string { + hash := strings.ToLower(strings.TrimSpace(torrent.Hash)) + if hash != "" { + return hash + } + parts := []string{torrent.Name, torrent.ContentPath, torrent.SavePath} + for i := range parts { + parts[i] = strings.TrimSpace(parts[i]) + } + key := strings.Join(parts, "|") + if strings.Trim(key, "|") == "" { + return "" + } + return strings.ToLower(key) +} + func (d *DownloadService) syncDownloadTaskProgress(ctx context.Context, torrent QBitTorrent, taskByKey map[string]model.DownloadTask) { if d == nil || d.repo == nil || d.repo.DB == nil || strings.TrimSpace(torrent.Name) == "" { return @@ -916,5 +1062,3 @@ func (d *DownloadService) completedTorrentSource(torrent QBitTorrent) string { } return "" } - - diff --git a/internal/service/downloads_test.go b/internal/service/downloads_test.go index 0d1415a..a53bf2c 100644 --- a/internal/service/downloads_test.go +++ b/internal/service/downloads_test.go @@ -87,6 +87,48 @@ func TestDownloadCompleteAutoOrganizesContentPath(t *testing.T) { } } +func TestDownloadPollBaselinesAlreadyCompletedTorrents(t *testing.T) { + repos := newOrganizerTestRepo(t) + svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil) + + svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{ + Hash: "already-complete", + Name: "Already Complete S01E01", + Progress: 1, + }}, nil) + + if got := len(svc.organizeQueue); got != 0 { + t.Fatalf("first poll queued %d organize jobs, want 0", got) + } + if !svc.prevStates["already-complete"] { + t.Fatal("first poll should remember completed baseline state") + } + + svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{ + Hash: "late-complete", + Name: "Late Complete S01E01", + Progress: 1, + }}, nil) + if got := len(svc.organizeQueue); got != 0 { + t.Fatalf("newly discovered completed torrent queued %d organize jobs, want 0", got) + } + + svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{ + Hash: "new-download", + Name: "New Download S01E01", + Progress: 0.5, + }}, nil) + svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{ + Hash: "new-download", + Name: "New Download S01E01", + Progress: 1, + }}, nil) + + if got := len(svc.organizeQueue); got != 1 { + t.Fatalf("completion transition queued %d organize jobs, want 1", got) + } +} + 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 != "测试影片" {