diff --git a/cmd/server/main.go b/cmd/server/main.go index 713f7a4..b3bf2bf 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -23,6 +23,7 @@ import ( "github.com/truewhile/MeBox/internal/config" "github.com/truewhile/MeBox/internal/database" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/repository" "github.com/truewhile/MeBox/internal/service" ) @@ -121,7 +122,7 @@ func main() { ) } }() - go services.Boot() + helper.Go(logger, "services.boot", services.Boot) // Graceful shutdown. stop := make(chan os.Signal, 1) diff --git a/internal/database/schema_migration.go b/internal/database/schema_migration.go index 896ae5c..ff6a393 100644 --- a/internal/database/schema_migration.go +++ b/internal/database/schema_migration.go @@ -8,6 +8,11 @@ import ( // AutoMigrate creates tables for every model registered in the model package. func AutoMigrate(db *gorm.DB) error { + // 必须先于 AutoMigrate:旧库中可能已有重复的 (user_id, media_id) 历史行, + // 不去重会导致唯一索引 uniq_user_history 创建失败。 + if err := dedupePlaybackHistories(db); err != nil { + return err + } if err := db.AutoMigrate(model.AllModels()...); err != nil { return err } @@ -37,6 +42,27 @@ func ensureSQLiteQueryOptimizer(db *gorm.DB) error { return db.Exec("ANALYZE").Error } +// dedupePlaybackHistories removes duplicate (user_id, media_id) rows left by +// the former read-then-write upsert, so the uniq_user_history composite unique +// index can be created on existing databases. Keeps the most recent row per +// pair, preferring live rows over soft-deleted ones. +func dedupePlaybackHistories(db *gorm.DB) error { + if !db.Migrator().HasTable("playback_histories") { + return nil + } + return db.Exec(` +DELETE FROM playback_histories WHERE id IN ( + SELECT id FROM ( + SELECT id, ROW_NUMBER() OVER ( + PARTITION BY user_id, media_id + ORDER BY deleted_at IS NULL DESC, watched_at DESC, id DESC + ) AS rn + FROM playback_histories + ) ranked + WHERE ranked.rn > 1 +)`).Error +} + func ensurePostgresColumnCompatibility(db *gorm.DB) error { if !isPostgres(db) { return nil diff --git a/internal/database/schema_migration_history_test.go b/internal/database/schema_migration_history_test.go new file mode 100644 index 0000000..7cba4ab --- /dev/null +++ b/internal/database/schema_migration_history_test.go @@ -0,0 +1,84 @@ +package database + +import ( + "testing" + "time" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" +) + +// TestAutoMigrateDedupesPlaybackHistories reproduces the upgrade path: a legacy +// database contains duplicate (user_id, media_id) history rows created by the +// old read-then-write upsert. AutoMigrate must merge them before creating the +// uniq_user_history composite unique index, otherwise the upgrade fails. +func TestAutoMigrateDedupesPlaybackHistories(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + // 旧 schema:无 uniq_user_history 唯一索引。 + if err := db.Exec(`CREATE TABLE playback_histories ( + id varchar(36) PRIMARY KEY, + created_at datetime, + updated_at datetime, + deleted_at datetime, + user_id varchar(36) NOT NULL, + media_id varchar(128) NOT NULL, + position_ms integer, + duration_ms integer, + watched_at datetime, + completed numeric + )`).Error; err != nil { + t.Fatal(err) + } + base := time.Now() + rows := []struct { + id string + position int64 + watchedAt time.Time + }{ + {"h-old", 1_000, base.Add(-2 * time.Hour)}, + {"h-mid", 2_000, base.Add(-1 * time.Hour)}, + {"h-new", 3_000, base}, + } + for _, r := range rows { + if err := db.Exec( + `INSERT INTO playback_histories (id, user_id, media_id, position_ms, watched_at, created_at, updated_at) + VALUES (?, 'u-1', 'm-1', ?, ?, ?, ?)`, + r.id, r.position, r.watchedAt, r.watchedAt, r.watchedAt, + ).Error; err != nil { + t.Fatal(err) + } + } + + if err := AutoMigrate(db); err != nil { + t.Fatalf("auto migrate with duplicate histories: %v", err) + } + + var count int64 + if err := db.Table("playback_histories").Where("user_id = ? AND media_id = ?", "u-1", "m-1").Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("expected duplicate rows merged to 1, got %d", count) + } + var position int64 + if err := db.Table("playback_histories"). + Where("user_id = ? AND media_id = ?", "u-1", "m-1"). + Select("position_ms").Scan(&position).Error; err != nil { + t.Fatal(err) + } + if position != 3_000 { + t.Fatalf("dedupe should keep the most recent row, got position_ms=%d", position) + } + + // 唯一索引存在时,重复插入同一 (user_id, media_id) 应触发冲突而非新增行。 + if err := db.Exec( + `INSERT INTO playback_histories (id, user_id, media_id, position_ms, watched_at, created_at, updated_at) + VALUES ('h-dup', 'u-1', 'm-1', 4_000, ?, ?, ?)`, + base, base, base, + ).Error; err == nil { + t.Fatal("insert violating uniq_user_history should fail") + } +} diff --git a/internal/handler/media.go b/internal/handler/media.go index 3c0d10a..afefcc1 100644 --- a/internal/handler/media.go +++ b/internal/handler/media.go @@ -11,6 +11,7 @@ import ( "github.com/gin-gonic/gin" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/middleware" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" @@ -164,18 +165,20 @@ func listLibrariesHandler(svc *service.Container) gin.HandlerFunc { case <-ctx.Done(): return } - acct := svc.EmbyRemote.AccountByID(ctx, v.AccountID) - if acct == nil { - return - } - tmpMount := &model.EmbyMount{Base: model.Base{ID: v.MountID}} - itemTypes := remoteLibraryItemTypes(v.CollectionType) - if _, total, err := svc.EmbyRemote.RemoteLibraryMedia(ctx, tmpMount, acct, v.RemoteID, itemTypes, 0, 1); err == nil { - remotePayloads[i].Total = total - } - if cards, err := svc.EmbyRemote.RemoteLatestCards(ctx, tmpMount, acct, v.RemoteID, limit); err == nil { - remotePayloads[i].Cards = cards - } + helper.Run(svc.Log, "media.remotePreview", func() { + acct := svc.EmbyRemote.AccountByID(ctx, v.AccountID) + if acct == nil { + return + } + tmpMount := &model.EmbyMount{Base: model.Base{ID: v.MountID}} + itemTypes := remoteLibraryItemTypes(v.CollectionType) + if _, total, err := svc.EmbyRemote.RemoteLibraryMedia(ctx, tmpMount, acct, v.RemoteID, itemTypes, 0, 1); err == nil { + remotePayloads[i].Total = total + } + if cards, err := svc.EmbyRemote.RemoteLatestCards(ctx, tmpMount, acct, v.RemoteID, limit); err == nil { + remotePayloads[i].Cards = cards + } + }) }() } wg.Wait() diff --git a/internal/handler/media_scan.go b/internal/handler/media_scan.go index 8c36d6f..ff7d04c 100644 --- a/internal/handler/media_scan.go +++ b/internal/handler/media_scan.go @@ -7,6 +7,7 @@ import ( "github.com/gin-gonic/gin" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/service" ) @@ -41,7 +42,11 @@ func scanLibraryHandler(svc *service.Container) gin.HandlerFunc { task := startScanHTTPTask(svc, "手动扫描入库", lib.Name, lib.Path) go func(libraryID string, task *service.TaskHandle, finish func()) { defer finish() - res, err := svc.Scan.ScanLibrary(context.Background(), libraryID) + var res *service.ScanResult + var err error + helper.Run(svc.Log, "scan.library", func() { + res, err = svc.Scan.ScanLibrary(context.Background(), libraryID) + }) if err != nil { finishHTTPTask(task, err, "scan", "手动扫描入库失败", scanTaskMetrics(res), scanTaskDetails(res, 20)) return @@ -75,7 +80,11 @@ func scanLibraryRootHandler(svc *service.Container) gin.HandlerFunc { task := startScanHTTPTask(svc, "手动扫描媒体库路径", id, rootID) go func(libraryID, libraryRootID string, task *service.TaskHandle, finish func()) { defer finish() - res, err := svc.Scan.ScanLibraryRoot(context.Background(), libraryID, libraryRootID) + var res *service.ScanResult + var err error + helper.Run(svc.Log, "scan.libraryRoot", func() { + res, err = svc.Scan.ScanLibraryRoot(context.Background(), libraryID, libraryRootID) + }) if err != nil { finishHTTPTask(task, err, "scan", "手动扫描路径失败", scanTaskMetrics(res), scanTaskDetails(res, 20)) return @@ -110,11 +119,13 @@ func queueLibraryRootScan(svc *service.Container, libraryID, rootID string) { } go func() { defer finish() - if strings.TrimSpace(rootID) == "" { - _, _ = svc.Scan.ScanLibrary(context.Background(), libraryID) - return - } - _, _ = svc.Scan.ScanLibraryRoot(context.Background(), libraryID, rootID) + helper.Run(svc.Log, "scan.queuedRoot", func() { + if strings.TrimSpace(rootID) == "" { + _, _ = svc.Scan.ScanLibrary(context.Background(), libraryID) + return + } + _, _ = svc.Scan.ScanLibraryRoot(context.Background(), libraryID, rootID) + }) }() } diff --git a/internal/helper/goroutine.go b/internal/helper/goroutine.go new file mode 100644 index 0000000..d948ddb --- /dev/null +++ b/internal/helper/goroutine.go @@ -0,0 +1,53 @@ +// Package helper provides shared utilities. +package helper + +import ( + "fmt" + "os" + "runtime/debug" + + "go.uber.org/zap" +) + +// Go runs fn in a new goroutine and recovers from panics so that a failure in +// a background task (scraper parsing remote responses, cloud-drive sync, ...) +// is logged instead of crashing the whole process. log may be nil. +func Go(log *zap.Logger, name string, fn func()) { + go Run(log, name, fn) +} + +// Run executes fn and recovers from panics, logging the task name and stack. +// Use it as the first statement inside goroutines spawned elsewhere, or wrap +// loop bodies so one bad iteration cannot kill a long-running worker. +func Run(log *zap.Logger, name string, fn func()) { + defer func() { + if r := recover(); r != nil { + logPanic(log, name, r) + } + }() + fn() +} + +// Recover runs fn and converts a panic into an error so callers can run their +// own deferred cleanup (releasing locks, updating job state) before unwinding. +func Recover(log *zap.Logger, name string, fn func() error) (err error) { + defer func() { + if r := recover(); r != nil { + logPanic(log, name, r) + err = fmt.Errorf("%s panicked: %v", name, r) + } + }() + return fn() +} + +func logPanic(log *zap.Logger, name string, r any) { + if log == nil { + fmt.Fprintf(os.Stderr, "background task panicked: task=%s panic=%v\n%s\n", name, r, debug.Stack()) + return + } + log.Error("background task panicked", + zap.String("task", name), + zap.Any("panic", r), + zap.ByteString("stack", debug.Stack()), + ) +} diff --git a/internal/helper/goroutine_test.go b/internal/helper/goroutine_test.go new file mode 100644 index 0000000..6d6b34e --- /dev/null +++ b/internal/helper/goroutine_test.go @@ -0,0 +1,95 @@ +package helper + +import ( + "errors" + "strings" + "testing" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" +) + +func newObservedLogger(t *testing.T) (*zap.Logger, *observer.ObservedLogs) { + t.Helper() + core, logs := observer.New(zap.ErrorLevel) + return zap.New(core), logs +} + +func waitForLogs(t *testing.T, logs *observer.ObservedLogs, n int) []observer.LoggedEntry { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if entries := logs.All(); len(entries) >= n { + return entries + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timed out waiting for %d log entries, got %d", n, logs.Len()) + return nil +} + +func TestRunRecoversPanic(t *testing.T) { + log, logs := newObservedLogger(t) + ran := false + Run(log, "unit.panic", func() { + ran = true + panic("boom") + }) + if !ran { + t.Fatal("fn should have run before panicking") + } + entries := waitForLogs(t, logs, 1) + if entries[0].Message != "background task panicked" { + t.Fatalf("unexpected message: %s", entries[0].Message) + } + found := false + for _, f := range entries[0].Context { + if f.Key == "task" && f.String == "unit.panic" { + found = true + } + } + if !found { + t.Fatalf("expected task name in log context: %v", entries[0].Context) + } +} + +func TestRunNoPanicNoLog(t *testing.T) { + log, logs := newObservedLogger(t) + Run(log, "unit.ok", func() {}) + time.Sleep(10 * time.Millisecond) + if logs.Len() != 0 { + t.Fatalf("expected no error log, got %d", logs.Len()) + } +} + +func TestRecoverConvertsPanicToError(t *testing.T) { + log, _ := newObservedLogger(t) + err := Recover(log, "unit.recover", func() error { + panic("kaboom") + }) + if err == nil { + t.Fatal("expected error from recovered panic") + } + if !strings.Contains(err.Error(), "kaboom") { + t.Fatalf("panic value should be in error: %v", err) + } +} + +func TestRecoverReturnsFnError(t *testing.T) { + sentinel := errors.New("plain failure") + err := Recover(nil, "unit.err", func() error { return sentinel }) + if !errors.Is(err, sentinel) { + t.Fatalf("expected fn error, got %v", err) + } +} + +func TestRunWithNilLoggerDoesNotCrash(t *testing.T) { + Run(nil, "unit.nillog", func() { panic("still caught") }) +} + +func TestGoLogsPanicFromSpawnedGoroutine(t *testing.T) { + log, logs := newObservedLogger(t) + Go(log, "unit.go", func() { panic("async boom") }) + waitForLogs(t, logs, 1) +} diff --git a/internal/model/playback_collection.go b/internal/model/playback_collection.go index 9979045..68ef3b2 100644 --- a/internal/model/playback_collection.go +++ b/internal/model/playback_collection.go @@ -3,10 +3,12 @@ package model import "time" // PlaybackHistory 记录当前播放位置以支持续播。 +// (user_id, media_id) 唯一:播放进度每几秒上报一次,唯一索引保证并发上报 +// 不会插入重复行(否则续播列表会出现重复卡片),也让 upsert 单语句完成。 type PlaybackHistory struct { Base - UserID string `gorm:"index;size:36;not null" json:"user_id"` - MediaID string `gorm:"index;size:128;not null" json:"media_id"` + UserID string `gorm:"index;size:36;not null;uniqueIndex:uniq_user_history" json:"user_id"` + MediaID string `gorm:"index;size:128;not null;uniqueIndex:uniq_user_history" json:"media_id"` PositionMs int64 `json:"position_ms"` DurationMs int64 `json:"duration_ms"` WatchedAt time.Time `json:"watched_at"` diff --git a/internal/repository/history_repository.go b/internal/repository/history_repository.go index d7c7314..5468ca7 100644 --- a/internal/repository/history_repository.go +++ b/internal/repository/history_repository.go @@ -2,9 +2,9 @@ package repository import ( "context" - "errors" "gorm.io/gorm" + "gorm.io/gorm/clause" "github.com/truewhile/MeBox/internal/model" ) @@ -13,25 +13,24 @@ import ( // upserts on (UserID, MediaID) so resume always reads the latest position. type HistoryRepository struct{ db *gorm.DB } -// Upsert atomically inserts/updates the resume position. +// Upsert atomically inserts/updates the resume position in a single statement, +// relying on the uniq_user_history composite unique index. Concurrent progress +// reports for the same (user, media) can no longer double-insert. func (r *HistoryRepository) Upsert(ctx context.Context, h *model.PlaybackHistory) error { - var existing model.PlaybackHistory - err := r.db.WithContext(ctx). - Where("user_id = ? AND media_id = ?", h.UserID, h.MediaID). - First(&existing).Error - if errors.Is(err, gorm.ErrRecordNotFound) { - return r.db.WithContext(ctx).Create(h).Error - } - if err != nil { - return err - } - existing.PositionMs = h.PositionMs - if h.DurationMs > 0 { - existing.DurationMs = h.DurationMs - } - existing.WatchedAt = h.WatchedAt - existing.Completed = h.Completed - return r.db.WithContext(ctx).Save(&existing).Error + return r.db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "user_id"}, {Name: "media_id"}}, + DoUpdates: clause.Assignments(map[string]any{ + "position_ms": h.PositionMs, + // 沿用旧语义:未知时长(0)不覆盖已记录的时长。 + "duration_ms": gorm.Expr( + "CASE WHEN ? > 0 THEN ? ELSE playback_histories.duration_ms END", + h.DurationMs, h.DurationMs, + ), + "watched_at": h.WatchedAt, + "completed": h.Completed, + "deleted_at": nil, + }), + }).Create(h).Error } // ListByUser returns the most recent history rows for the user. diff --git a/internal/repository/history_repository_test.go b/internal/repository/history_repository_test.go new file mode 100644 index 0000000..7a09c34 --- /dev/null +++ b/internal/repository/history_repository_test.go @@ -0,0 +1,79 @@ +package repository + +import ( + "testing" + "time" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" + + "github.com/truewhile/MeBox/internal/database" + "github.com/truewhile/MeBox/internal/model" +) + +func TestHistoryUpsertSingleRowPerUserMedia(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := database.AutoMigrate(db); err != nil { + t.Fatalf("migrate: %v", err) + } + repos := New(db) + ctx := t.Context() + watched := time.Now() + + first := &model.PlaybackHistory{UserID: "u-1", MediaID: "m-1", PositionMs: 30_000, DurationMs: 0, WatchedAt: watched, Completed: false} + if err := repos.History.Upsert(ctx, first); err != nil { + t.Fatalf("first upsert: %v", err) + } + second := &model.PlaybackHistory{UserID: "u-1", MediaID: "m-1", PositionMs: 90_000, DurationMs: 120_000, WatchedAt: watched.Add(time.Minute), Completed: true} + if err := repos.History.Upsert(ctx, second); err != nil { + t.Fatalf("second upsert: %v", err) + } + + var count int64 + if err := db.Model(&model.PlaybackHistory{}).Where("user_id = ? AND media_id = ?", "u-1", "m-1").Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("expected 1 history row after upserts, got %d", count) + } + var got model.PlaybackHistory + if err := db.Where("user_id = ? AND media_id = ?", "u-1", "m-1").First(&got).Error; err != nil { + t.Fatal(err) + } + if got.PositionMs != 90_000 || !got.Completed { + t.Fatalf("position/completion not updated: %#v", got) + } + if got.DurationMs != 120_000 { + t.Fatalf("duration should update when known, got %d", got.DurationMs) + } +} + +func TestHistoryUpsertKeepsDurationWhenUnknown(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := database.AutoMigrate(db); err != nil { + t.Fatalf("migrate: %v", err) + } + repos := New(db) + ctx := t.Context() + watched := time.Now() + + if err := repos.History.Upsert(ctx, &model.PlaybackHistory{UserID: "u-1", MediaID: "m-2", PositionMs: 10, DurationMs: 600_000, WatchedAt: watched}); err != nil { + t.Fatal(err) + } + if err := repos.History.Upsert(ctx, &model.PlaybackHistory{UserID: "u-1", MediaID: "m-2", PositionMs: 20, DurationMs: 0, WatchedAt: watched.Add(time.Second)}); err != nil { + t.Fatal(err) + } + var got model.PlaybackHistory + if err := db.Where("user_id = ? AND media_id = ?", "u-1", "m-2").First(&got).Error; err != nil { + t.Fatal(err) + } + if got.DurationMs != 600_000 { + t.Fatalf("duration_ms=0 upsert must not clear stored duration, got %d", got.DurationMs) + } +} diff --git a/internal/service/audit.go b/internal/service/audit.go index 8fc30b4..9ca88a8 100644 --- a/internal/service/audit.go +++ b/internal/service/audit.go @@ -12,6 +12,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" ) @@ -50,6 +51,8 @@ func (a *AuditService) RecordBestEffort(userID, action, target, ip, detail strin go func() { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - a.Record(ctx, userID, action, target, ip, detail) + helper.Run(a.log, "audit.record", func() { + a.Record(ctx, userID, action, target, ip, detail) + }) }() } diff --git a/internal/service/auth.go b/internal/service/auth.go index 0048f3e..a8ac532 100644 --- a/internal/service/auth.go +++ b/internal/service/auth.go @@ -13,6 +13,7 @@ import ( "golang.org/x/crypto/bcrypt" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" ) @@ -176,9 +177,11 @@ func (s *AuthService) touchLoginBestEffort(userID string) { 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)) - } + helper.Run(s.log, "auth.touchLogin", func() { + 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)) + } + }) }() } diff --git a/internal/service/emby_counts.go b/internal/service/emby_counts.go index 764bc66..a1f0695 100644 --- a/internal/service/emby_counts.go +++ b/internal/service/emby_counts.go @@ -54,7 +54,7 @@ func (e *EmbyService) countVisibleSeries(ctx context.Context, userID string) (in for i := range rows { key := strings.TrimSpace(rows[i].SeriesID) if key == "" { - key = stableEmbyID(embyVirtualSeriesPrefix, rows[i].LibraryID, e.seriesNameForMedia(&rows[i])) + key = stableEmbyID(embyVirtualSeriesPrefix, rows[i].LibraryID, e.seriesNameForMedia(ctx, &rows[i])) } seen[key] = struct{}{} } diff --git a/internal/service/emby_items_detail.go b/internal/service/emby_items_detail.go index 0e5b57d..0ef7f75 100644 --- a/internal/service/emby_items_detail.go +++ b/internal/service/emby_items_detail.go @@ -98,7 +98,8 @@ func (e *EmbyService) Item(ctx context.Context, mediaID, userID string) (map[str pos = h.PositionMs } } - return e.itemPayload(ctx, m, fav, pos), nil + // 单条目 payload 内部对库类型/series 标题有多次查找,挂请求级缓存合并。 + return e.itemPayload(e.withPayloadCache(ctx), m, fav, pos), nil } // LatestItems 最近添加,全库或指定库。远程媒体库(parentID 带前缀)直接透传远程。 @@ -177,7 +178,7 @@ func (e *EmbyService) latestSeriesItemsForLibrary(ctx context.Context, userID, l if err := q.Order(mediaReleaseOrderSQL(true)).Limit(embySeriesGroupingLimit).Find(&rows).Error; err != nil { return nil, err } - groups := e.seriesGroupsFromMedia(rows) + groups := e.seriesGroupsFromMedia(ctx, rows) sortSeriesGroups(groups, ItemsParams{SortBy: "premieredate", SortOrder: "Descending"}) if len(groups) > limit { groups = groups[:limit] @@ -415,9 +416,9 @@ func (e *EmbyService) itemPayload(ctx context.Context, m *model.Media, fav bool, seasonID := "" if e.mediaShouldBeEpisode(ctx, m) { itemType = "Episode" - seriesID = e.seriesIDForMedia(m) - seriesName = e.seriesNameForMedia(m) - seasonID = e.seasonIDForMedia(m) + seriesID = e.seriesIDForMedia(ctx, m) + seriesName = e.seriesNameForMedia(ctx, m) + seasonID = e.seasonIDForMedia(ctx, m) parentID = seasonID episodeTitle := strings.TrimSpace(m.EpisodeTitle) if episodeTitle != "" { diff --git a/internal/service/emby_items_list.go b/internal/service/emby_items_list.go index fc60aef..88fe491 100644 --- a/internal/service/emby_items_list.go +++ b/internal/service/emby_items_list.go @@ -149,6 +149,9 @@ func (e *EmbyService) episodeItems(ctx context.Context, rows []model.Media, p It } func (e *EmbyService) payloadsForMedia(ctx context.Context, rows []model.Media, userID string) ([]map[string]any, error) { + // 请求级缓存:库类型与 series 标题整页只查一次,消除逐条目 N+1。 + ctx = e.withPayloadCache(ctx) + e.prefetchPayloadCache(ctx, rows) rows = e.collapseMediaVersionRows(ctx, rows) userFavs := map[string]bool{} userPos := map[string]int64{} @@ -240,7 +243,7 @@ func (e *EmbyService) seriesItemsForLibrary(ctx context.Context, libraryID strin if err := q.Order(mediaReleaseOrderSQL(true)).Limit(embySeriesGroupingLimit).Find(&rows).Error; err != nil { return nil, err } - groups := e.seriesGroupsFromMedia(rows) + groups := e.seriesGroupsFromMedia(ctx, rows) sortSeriesGroups(groups, p) total := len(groups) items := make([]map[string]any, 0, minInt(p.Limit, len(groups))) diff --git a/internal/service/emby_movie_items.go b/internal/service/emby_movie_items.go index 5337945..2257764 100644 --- a/internal/service/emby_movie_items.go +++ b/internal/service/emby_movie_items.go @@ -65,7 +65,7 @@ func (e *EmbyService) movieLibraryItems(ctx context.Context, p ItemsParams) (map return nil, err } } - seriesGroups := e.seriesGroupsFromMedia(episodicRows) + seriesGroups := e.seriesGroupsFromMedia(ctx, episodicRows) // 真正的电影 -> Movie 项(剔除剧集结构行)。 movieQ := apply(e.repo.DB.WithContext(ctx).Model(&model.Media{})) @@ -135,10 +135,11 @@ func (e *EmbyService) libraryIsEpisodic(ctx context.Context, libraryID string) ( if strings.TrimSpace(libraryID) == "" { return false, nil } - if lib, err := e.repo.Library.FindByID(ctx, libraryID); err != nil { + // 走请求级缓存(若有),避免同一请求内对同一库重复查表。 + if typ, ok, err := e.payloadLibraryType(ctx, libraryID); err != nil { return false, err - } else if lib != nil { - return embyLibraryTypeIsEpisodic(lib.Type), nil + } else if ok { + return embyLibraryTypeIsEpisodic(typ), nil } var count int64 err := e.repo.DB.WithContext(ctx).Model(&model.Media{}). @@ -151,11 +152,11 @@ func (e *EmbyService) mediaBelongsToEpisodicLibrary(ctx context.Context, m *mode if e == nil || m == nil || strings.TrimSpace(m.LibraryID) == "" { return false } - lib, err := e.repo.Library.FindByID(ctx, m.LibraryID) - if err != nil || lib == nil { + typ, ok, err := e.payloadLibraryType(ctx, m.LibraryID) + if err != nil || !ok { return false } - return embyLibraryTypeIsEpisodic(lib.Type) + return embyLibraryTypeIsEpisodic(typ) } func (e *EmbyService) mediaShouldBeEpisode(ctx context.Context, m *model.Media) bool { diff --git a/internal/service/emby_payload_ctx.go b/internal/service/emby_payload_ctx.go new file mode 100644 index 0000000..9bb7068 --- /dev/null +++ b/internal/service/emby_payload_ctx.go @@ -0,0 +1,175 @@ +package service + +import ( + "context" + "errors" + "strings" + "sync" + + "gorm.io/gorm" + + "github.com/truewhile/MeBox/internal/model" +) + +// 请求级 payload 构建缓存:/Items 列表为每行构建 payload 时, +// mediaShouldBeEpisode 需要库类型、剧集 payload 需要 series 标题。 +// 一次页面请求内这些值高度重复(同一库、同一部剧),挂在 ctx 上的 +// 小缓存可以把每条目 2-3 次 DB 查询降为整个请求各 1 次预取。 + +type embyPayloadCacheKey struct{} + +type embyLibraryTypeEntry struct { + typ string + found bool // 库不存在时 found=false,调用方可退回计数启发式 +} + +type embyPayloadCache struct { + mu sync.Mutex + libTypes map[string]embyLibraryTypeEntry + series map[string]string // series_id -> title("" 表示不存在/无标题) +} + +func (c *embyPayloadCache) libraryType(id string) (embyLibraryTypeEntry, bool) { + c.mu.Lock() + defer c.mu.Unlock() + entry, ok := c.libTypes[id] + return entry, ok +} + +func (c *embyPayloadCache) setLibraryType(id string, entry embyLibraryTypeEntry) { + c.mu.Lock() + defer c.mu.Unlock() + c.libTypes[id] = entry +} + +func (c *embyPayloadCache) seriesTitle(id string) (string, bool) { + c.mu.Lock() + defer c.mu.Unlock() + title, ok := c.series[id] + return title, ok +} + +func (c *embyPayloadCache) setSeriesTitle(id, title string) { + c.mu.Lock() + defer c.mu.Unlock() + c.series[id] = title +} + +// withPayloadCache attaches a fresh request-scoped cache if none exists yet. +func (e *EmbyService) withPayloadCache(ctx context.Context) context.Context { + if e == nil || e.repo == nil { + return ctx + } + if ctx.Value(embyPayloadCacheKey{}) != nil { + return ctx + } + return context.WithValue(ctx, embyPayloadCacheKey{}, &embyPayloadCache{ + libTypes: map[string]embyLibraryTypeEntry{}, + series: map[string]string{}, + }) +} + +// prefetchPayloadCache warms the cache for the given media rows with two bulk +// queries (library types, series titles) instead of per-item lookups. +func (e *EmbyService) prefetchPayloadCache(ctx context.Context, rows []model.Media) { + cache, ok := ctx.Value(embyPayloadCacheKey{}).(*embyPayloadCache) + if !ok || len(rows) == 0 { + return + } + libIDs := make([]string, 0, 8) + seriesIDs := make([]string, 0, 8) + seenLib := map[string]struct{}{} + seenSeries := map[string]struct{}{} + for i := range rows { + row := &rows[i] + if id := strings.TrimSpace(row.LibraryID); id != "" { + if _, done := seenLib[id]; !done { + // 已在缓存中的库不必再查。 + if _, hit := cache.libraryType(id); !hit { + seenLib[id] = struct{}{} + libIDs = append(libIDs, id) + } + } + } + if id := strings.TrimSpace(row.SeriesID); id != "" { + if _, done := seenSeries[id]; !done { + if _, hit := cache.seriesTitle(id); !hit { + seenSeries[id] = struct{}{} + seriesIDs = append(seriesIDs, id) + } + } + } + } + if len(libIDs) > 0 { + var libs []model.Library + if err := e.repo.DB.WithContext(ctx).Select("id, type").Where("id IN ?", libIDs).Find(&libs).Error; err == nil { + found := map[string]string{} + for _, lib := range libs { + found[lib.ID] = lib.Type + } + for _, id := range libIDs { + typ, ok := found[id] + cache.setLibraryType(id, embyLibraryTypeEntry{typ: typ, found: ok}) + } + } + } + if len(seriesIDs) > 0 { + var series []model.Series + if err := e.repo.DB.WithContext(ctx).Select("id, title").Where("id IN ?", seriesIDs).Find(&series).Error; err == nil { + for _, s := range series { + cache.setSeriesTitle(s.ID, s.Title) + } + } + } +} + +// payloadLibraryType resolves a library type through the request cache, +// falling back to a direct lookup when no cache is attached. found=false +// means the library row does not exist (soft-deleted or orphaned id). +func (e *EmbyService) payloadLibraryType(ctx context.Context, libraryID string) (typ string, found bool, err error) { + if cache, ok := ctx.Value(embyPayloadCacheKey{}).(*embyPayloadCache); ok { + if entry, hit := cache.libraryType(libraryID); hit { + return entry.typ, entry.found, nil + } + var lib model.Library + if dbErr := e.repo.DB.WithContext(ctx).Select("id, type").Where("id = ?", libraryID).First(&lib).Error; dbErr != nil { + cache.setLibraryType(libraryID, embyLibraryTypeEntry{}) + return "", false, nil + } + cache.setLibraryType(lib.ID, embyLibraryTypeEntry{typ: lib.Type, found: true}) + return lib.Type, true, nil + } + var lib model.Library + if err = e.repo.DB.WithContext(ctx).Select("id, type").Where("id = ?", libraryID).First(&lib).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return "", false, nil + } + return "", false, err + } + return lib.Type, true, nil +} + +// payloadSeriesTitle resolves a series title through the request cache, +// falling back to a direct lookup when no cache is attached. +func (e *EmbyService) payloadSeriesTitle(ctx context.Context, seriesID string) (string, bool, error) { + if cache, ok := ctx.Value(embyPayloadCacheKey{}).(*embyPayloadCache); ok { + if title, hit := cache.seriesTitle(seriesID); hit { + return title, true, nil + } + var s model.Series + if err := e.repo.DB.WithContext(ctx).Select("id, title").Where("id = ?", seriesID).First(&s).Error; err != nil { + cache.setSeriesTitle(seriesID, "") + return "", true, nil + } + cache.setSeriesTitle(s.ID, s.Title) + return s.Title, true, nil + } + series, err := e.repo.Series.FindByID(ctx, seriesID) + if err != nil { + return "", false, err + } + if series == nil { + return "", false, nil + } + return series.Title, true, nil +} diff --git a/internal/service/emby_remote_web.go b/internal/service/emby_remote_web.go index 27f1584..57e57e4 100644 --- a/internal/service/emby_remote_web.go +++ b/internal/service/emby_remote_web.go @@ -18,6 +18,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" ) @@ -609,32 +610,34 @@ func (r *EmbyRemoteService) RemoteSearchMedia(ctx context.Context, query string, return } - q := url.Values{} - q.Set("ParentId", target.mount.RemoteViewID) - q.Set("Recursive", "true") - q.Set("SearchTerm", query) - q.Set("IncludeItemTypes", "Movie,Series") - q.Set("Fields", "Overview,Genres,ProviderIds,Path,SeriesPrimaryImage,MediaStreams,MediaSources,DateCreated,PremiereDate,ProductionYear,CommunityRating,CriticRating") - q.Set("Limit", strconv.Itoa(limit)) - q.Set("StartIndex", "0") + helper.Run(r.log, "emby.remoteSearch", func() { + q := url.Values{} + q.Set("ParentId", target.mount.RemoteViewID) + q.Set("Recursive", "true") + q.Set("SearchTerm", query) + q.Set("IncludeItemTypes", "Movie,Series") + q.Set("Fields", "Overview,Genres,ProviderIds,Path,SeriesPrimaryImage,MediaStreams,MediaSources,DateCreated,PremiereDate,ProductionYear,CommunityRating,CriticRating") + q.Set("Limit", strconv.Itoa(limit)) + q.Set("StartIndex", "0") - var body struct { - Items []map[string]any `json:"Items"` - } - if err := r.doGet(searchCtx, target.acct, target.cfg, "/Users/"+url.PathEscape(r.remoteUserID(target.cfg))+"/Items", q, &body); err != nil { - if r.log != nil { - r.log.Warn("remote search failed", - zap.String("mount", target.mount.RemoteViewName), zap.Error(err)) + var body struct { + Items []map[string]any `json:"Items"` } - return - } - medias := make([]model.Media, 0, len(body.Items)) - for _, it := range body.Items { - RewriteEmbyRemoteIDs(it, target.mount.ID) - m := r.MapRemoteItemToMedia(searchCtx, &target.mount, target.acct, target.cfg, it) - medias = append(medias, m) - } - results[idx] = searchResult{items: medias} + if err := r.doGet(searchCtx, target.acct, target.cfg, "/Users/"+url.PathEscape(r.remoteUserID(target.cfg))+"/Items", q, &body); err != nil { + if r.log != nil { + r.log.Warn("remote search failed", + zap.String("mount", target.mount.RemoteViewName), zap.Error(err)) + } + return + } + medias := make([]model.Media, 0, len(body.Items)) + for _, it := range body.Items { + RewriteEmbyRemoteIDs(it, target.mount.ID) + m := r.MapRemoteItemToMedia(searchCtx, &target.mount, target.acct, target.cfg, it) + medias = append(medias, m) + } + results[idx] = searchResult{items: medias} + }) }(i, t) } wg.Wait() diff --git a/internal/service/emby_series.go b/internal/service/emby_series.go index 6b67428..c473c16 100644 --- a/internal/service/emby_series.go +++ b/internal/service/emby_series.go @@ -49,11 +49,15 @@ func (e *EmbyService) findSeriesGroup(ctx context.Context, id, userID string) (e q = e.applyUserMediaVisibility(ctx, q, userID) if !strings.HasPrefix(id, embyVirtualSeriesPrefix) { q = q.Where("series_id = ?", id) + } else { + // 虚拟 series ID 只可能来自 series_id 为空的媒体: + // 有 series_id 时分组 key 就是 series_id 本身(UUID,不带虚拟前缀)。 + q = q.Where("series_id IS NULL OR series_id = ''") } if err := q.Order("media.season_num asc, media.episode_num asc, media.created_at asc").Limit(embySeriesGroupingLimit).Find(&rows).Error; err != nil { return embySeriesGroup{}, false, err } - for _, group := range e.seriesGroupsFromMedia(rows) { + for _, group := range e.seriesGroupsFromMedia(ctx, rows) { if group.ID == id { e.rememberSeriesGroup(group) return group, true, nil @@ -88,9 +92,18 @@ func (e *EmbyService) findSeasonGroup(ctx context.Context, id, userID string) (e if season, ok := e.cachedSeasonGroup(id); ok { return season, true, nil } + // 虚拟 Season ID 是 hash(seriesKey, seasonNum),无法反解出 series。 + // 常见情况(已刮削、series_id 非空)先用一条小型 DISTINCT 查询枚举候选对, + // 在内存中算哈希匹配,命中后只加载该一部剧的剧集行,避免整库扫描。 + if season, ok, err := e.findSeasonGroupBySeriesCandidates(ctx, id, userID); err != nil { + return embySeasonGroup{}, false, err + } else if ok { + return season, true, nil + } + // 回退:未刮削(series_id 为空,虚拟 key 由库名+名称派生)的媒体只能全量分组。 var rows []model.Media q := e.repo.DB.WithContext(ctx).Model(&model.Media{}). - Where("season_num > 0 OR episode_num > 0") + Where("(series_id IS NULL OR series_id = '') AND (season_num > 0 OR episode_num > 0)") q = e.applyUserMediaVisibility(ctx, q, userID) if err := q. Order("media.season_num asc, media.episode_num asc, media.created_at asc"). @@ -98,7 +111,7 @@ func (e *EmbyService) findSeasonGroup(ctx context.Context, id, userID string) (e Find(&rows).Error; err != nil { return embySeasonGroup{}, false, err } - for _, series := range e.seriesGroupsFromMedia(rows) { + for _, series := range e.seriesGroupsFromMedia(ctx, rows) { for _, season := range e.seasonsForSeries(series) { if season.ID == id { e.rememberSeriesGroup(series) @@ -109,18 +122,77 @@ func (e *EmbyService) findSeasonGroup(ctx context.Context, id, userID string) (e return embySeasonGroup{}, false, nil } -func (e *EmbyService) seriesGroupsFromMedia(rows []model.Media) []embySeriesGroup { +// findSeasonGroupBySeriesCandidates resolves virtual season IDs for media that +// carry a real series_id: enumerate distinct (series_id, season_num) pairs via +// SQL, hash each candidate to find the matching season, then load only that +// one series' episodes. +func (e *EmbyService) findSeasonGroupBySeriesCandidates(ctx context.Context, id, userID string) (embySeasonGroup, bool, error) { + type seasonCandidate struct { + SeriesID string + SeasonNum int + } + var candidates []seasonCandidate + q := e.repo.DB.WithContext(ctx).Model(&model.Media{}). + Select("DISTINCT series_id, season_num"). + Where("series_id <> '' AND (season_num > 0 OR episode_num > 0)") + q = e.applyUserMediaVisibility(ctx, q, userID) + if err := q.Find(&candidates).Error; err != nil { + return embySeasonGroup{}, false, err + } + matched := make([]string, 0, 1) + for _, cand := range candidates { + if seasonID(cand.SeriesID, cand.SeasonNum) == id { + matched = append(matched, cand.SeriesID) + } + } + for _, matchedSeries := range matched { + season, ok, err := e.seasonGroupForSeries(ctx, id, matchedSeries, userID) + if err != nil || ok { + return season, ok, err + } + } + return embySeasonGroup{}, false, nil +} + +// seasonGroupForSeries rebuilds the season groups of one series (small row +// set) and returns the one matching the virtual season id. +func (e *EmbyService) seasonGroupForSeries(ctx context.Context, id, seriesID, userID string) (embySeasonGroup, bool, error) { + var rows []model.Media + rq := e.repo.DB.WithContext(ctx).Model(&model.Media{}). + Where("series_id = ? AND (season_num > 0 OR episode_num > 0)", seriesID) + rq = e.applyUserMediaVisibility(ctx, rq, userID) + if err := rq. + Order("media.season_num asc, media.episode_num asc, media.created_at asc"). + Limit(embySeriesGroupingLimit). + Find(&rows).Error; err != nil { + return embySeasonGroup{}, false, err + } + for _, series := range e.seriesGroupsFromMedia(ctx, rows) { + if series.ID != seriesID { + continue + } + for _, season := range e.seasonsForSeries(series) { + if season.ID == id { + e.rememberSeriesGroup(series) + return season, true, nil + } + } + } + return embySeasonGroup{}, false, nil +} + +func (e *EmbyService) seriesGroupsFromMedia(ctx context.Context, rows []model.Media) []embySeriesGroup { byID := map[string]*embySeriesGroup{} order := []string{} for _, row := range rows { row := row - seriesID := e.seriesIDForMedia(&row) + seriesID := e.seriesIDForMedia(ctx, &row) group, ok := byID[seriesID] if !ok { group = &embySeriesGroup{ ID: seriesID, LibraryID: row.LibraryID, - Name: e.seriesNameForMedia(&row), + Name: e.seriesNameForMedia(ctx, &row), Year: row.Year, ReleaseDate: row.ReleaseDate, TMDbID: row.TMDbID, diff --git a/internal/service/emby_series_ids.go b/internal/service/emby_series_ids.go index 3c9a8c6..43ff625 100644 --- a/internal/service/emby_series_ids.go +++ b/internal/service/emby_series_ids.go @@ -14,21 +14,22 @@ import ( "github.com/truewhile/MeBox/internal/model" ) -func (e *EmbyService) seriesIDForMedia(m *model.Media) string { +func (e *EmbyService) seriesIDForMedia(ctx context.Context, m *model.Media) string { if strings.TrimSpace(m.SeriesID) != "" { return m.SeriesID } - return stableEmbyID(embyVirtualSeriesPrefix, m.LibraryID, e.seriesNameForMedia(m)) + return stableEmbyID(embyVirtualSeriesPrefix, m.LibraryID, e.seriesNameForMedia(ctx, m)) } -func (e *EmbyService) seasonIDForMedia(m *model.Media) string { - return seasonID(e.seriesIDForMedia(m), m.SeasonNum) +func (e *EmbyService) seasonIDForMedia(ctx context.Context, m *model.Media) string { + return seasonID(e.seriesIDForMedia(ctx, m), m.SeasonNum) } -func (e *EmbyService) seriesNameForMedia(m *model.Media) string { +func (e *EmbyService) seriesNameForMedia(ctx context.Context, m *model.Media) string { if strings.TrimSpace(m.SeriesID) != "" { - if series, err := e.repo.Series.FindByID(context.Background(), m.SeriesID); err == nil && series != nil && strings.TrimSpace(series.Title) != "" { - return series.Title + // 走请求级缓存;无缓存 ctx 时退化为单次查询。 + if title, ok, err := e.payloadSeriesTitle(ctx, m.SeriesID); err == nil && ok && strings.TrimSpace(title) != "" { + return title } } if strings.EqualFold(strings.TrimSpace(m.ScrapeStatus), "matched") && strings.TrimSpace(m.Title) != "" { diff --git a/internal/service/ffmpeg_tools.go b/internal/service/ffmpeg_tools.go index 056bde0..f1ab29f 100644 --- a/internal/service/ffmpeg_tools.go +++ b/internal/service/ffmpeg_tools.go @@ -21,6 +21,7 @@ import ( "go.uber.org/zap" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/repository" ) @@ -152,7 +153,7 @@ func (s *FFmpegToolsService) StartInstall(ctx context.Context) error { s.mu.Unlock() s.setMessage("准备下载…") - go s.runInstall() + helper.Go(s.log, "ffmpeg.install", s.runInstall) return nil } diff --git a/internal/service/scanner_local_probe_queue.go b/internal/service/scanner_local_probe_queue.go index c4e2018..a2b4422 100644 --- a/internal/service/scanner_local_probe_queue.go +++ b/internal/service/scanner_local_probe_queue.go @@ -7,6 +7,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" ) @@ -45,7 +46,7 @@ func (s *ScannerService) startLocalMediaProbeWorkers() { s.localMediaProbeOnce.Do(func() { workers := s.ffprobeWorkerCount() for i := 0; i < workers; i++ { - go s.localMediaProbeWorker() + helper.Go(s.log, "scanner.probeWorker", s.localMediaProbeWorker) } }) } diff --git a/internal/service/scanner_post_scan.go b/internal/service/scanner_post_scan.go index 9719734..608246e 100644 --- a/internal/service/scanner_post_scan.go +++ b/internal/service/scanner_post_scan.go @@ -5,6 +5,8 @@ import ( "time" "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/helper" ) func (s *ScannerService) invalidateMediaCache(ctx context.Context) { @@ -18,10 +20,12 @@ func (s *ScannerService) startAutoScrape(ctx context.Context, libraryID string) scrapeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Minute) go func() { defer cancel() - _, err := s.scraper.EnrichLibraryDetailedWithOptions(scrapeCtx, libraryID, skipEpisodeArtworkOptions(false)) - if err != nil { - s.log.Warn("scraper enrich failed", zap.Error(err)) - return - } + // 扫描触发的后台刮削与请求线程无关,panic 只记日志,不能带崩进程。 + helper.Run(s.log, "scanner.autoScrape", func() { + _, err := s.scraper.EnrichLibraryDetailedWithOptions(scrapeCtx, libraryID, skipEpisodeArtworkOptions(false)) + if err != nil { + s.log.Warn("scraper enrich failed", zap.Error(err)) + } + }) }() } diff --git a/internal/service/scheduler.go b/internal/service/scheduler.go index a8a28f4..9974f7a 100644 --- a/internal/service/scheduler.go +++ b/internal/service/scheduler.go @@ -22,6 +22,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/repository" ) @@ -139,7 +140,7 @@ func (s *SchedulerService) Start(ctx context.Context) { // 首轮等满一个完整周期再跑,平时节奏不变。 initialDelay = j.interval } - go s.loopWithInitialDelay(ctx, j, initialDelay) + helper.Go(s.log, "scheduler.loop."+j.name, func() { s.loopWithInitialDelay(ctx, j, initialDelay) }) } } diff --git a/internal/service/scheduler_runner.go b/internal/service/scheduler_runner.go index 884cac8..ab6ec23 100644 --- a/internal/service/scheduler_runner.go +++ b/internal/service/scheduler_runner.go @@ -6,6 +6,8 @@ import ( "time" "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/helper" ) // JobStatus is a snapshot suitable for the admin UI. @@ -142,7 +144,8 @@ func (s *SchedulerService) beginRun(j *scheduledJob) error { } func (s *SchedulerService) runReserved(ctx context.Context, j *scheduledJob) error { - err := j.run(ctx) + // 任务 panic 转为 error,保证下方 running/lastErr 状态照常清理、调度循环存活。 + err := helper.Recover(s.log, "scheduler.job."+j.name, func() error { return j.run(ctx) }) s.mu.Lock() j.lastRun = s.currentTime() if err != nil { diff --git a/internal/service/scraper_queue.go b/internal/service/scraper_queue.go index 04408c7..a63946c 100644 --- a/internal/service/scraper_queue.go +++ b/internal/service/scraper_queue.go @@ -10,6 +10,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" ) @@ -73,7 +74,10 @@ func (s *ScraperService) queueWorker(ctx context.Context) { } defer func() { <-sem }() - s.processScrapeTask(ctx, t) + // 刮削要解析远端元数据响应,单个任务 panic 不应拖垮队列 worker。 + helper.Run(s.log, "scraper.task", func() { + s.processScrapeTask(ctx, t) + }) }(&tasks[i]) } wg.Wait() diff --git a/internal/service/service.go b/internal/service/service.go index 6a631c5..3f4f6d4 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -10,6 +10,7 @@ import ( "go.uber.org/zap" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/repository" ) @@ -96,7 +97,7 @@ func (c *Container) Boot() { if err := c.APIConfig.SeedDefaults(c.stopCtx); err != nil { c.Log.Warn("api config seed failed", zap.Error(err)) } - go c.warmMediaSearchIndex(c.stopCtx) + helper.Go(c.Log, "service.warmMediaSearchIndex", func() { c.warmMediaSearchIndex(c.stopCtx) }) // 启动调度器定时任务 c.Scheduler.Start(c.stopCtx) @@ -119,7 +120,7 @@ func (c *Container) Boot() { // Mgo 保号规则巡检:默认关闭,由管理员通过 Telegram Bot 命令开启。 // 每天触发一次评估;规则里的窗口可随机,不固定。 if c.Device != nil { - go c.runInactivitySweeper(c.stopCtx) + helper.Go(c.Log, "service.inactivitySweeper", func() { c.runInactivitySweeper(c.stopCtx) }) } } diff --git a/internal/service/service_builder.go b/internal/service/service_builder.go index 907158a..37248a7 100644 --- a/internal/service/service_builder.go +++ b/internal/service/service_builder.go @@ -8,6 +8,7 @@ import ( "go.uber.org/zap" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" ) @@ -47,12 +48,12 @@ func newServiceContainer(cfg *config.Config, log *zap.Logger, repos *repository. func (b *serviceContainerBuilder) startRealtimeServices() { b.c.WSHub = NewHub(b.log) - go b.c.WSHub.Run() + helper.Go(b.log, "ws.hub", b.c.WSHub.Run) b.c.Tasks = NewTaskTrackerService(b.log, b.c.WSHub) b.c.SystemUpdate = NewSystemUpdateService(b.cfg, b.log, b.repos, b.c.Tasks, b.version) b.c.SSEHub = NewSSEHub(b.log) - go b.c.SSEHub.Run() + helper.Go(b.log, "sse.hub", b.c.SSEHub.Run) } func (b *serviceContainerBuilder) initProviderServices() { diff --git a/internal/service/strm_queue.go b/internal/service/strm_queue.go index fe5fac1..4fd51ec 100644 --- a/internal/service/strm_queue.go +++ b/internal/service/strm_queue.go @@ -18,6 +18,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/service/cloud" "github.com/truewhile/MeBox/internal/service/cloud115" @@ -66,7 +67,10 @@ func (s *StrmService) downloadWorker(ctx context.Context) { return } defer s.releaseDownloadSlot(task.Provider) - s.processDownloadTask(ctx, task) + // 单个任务 panic 不应拖垮整个下载 worker。 + helper.Run(s.log, "strm.downloadTask", func() { + s.processDownloadTask(ctx, task) + }) }(i) } wg.Wait() diff --git a/internal/service/strm_service.go b/internal/service/strm_service.go index d04e43c..43a099c 100644 --- a/internal/service/strm_service.go +++ b/internal/service/strm_service.go @@ -24,6 +24,7 @@ import ( "go.uber.org/zap" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" "github.com/truewhile/MeBox/internal/service/cloud" @@ -186,14 +187,14 @@ func (s *StrmService) Start(ctx context.Context) { uploadThreads = 4 } for i := 0; i < downloadThreads; i++ { - go s.downloadWorker(ctx) + helper.Go(s.log, "strm.downloadWorker", func() { s.downloadWorker(ctx) }) } for i := 0; i < uploadThreads; i++ { - go s.uploadWorker(ctx) + helper.Go(s.log, "strm.uploadWorker", func() { s.uploadWorker(ctx) }) } - go s.cronLoop(ctx) - go s.queueCleanupLoop(ctx) - go s.refresh115TokensLoop(ctx) + helper.Go(s.log, "strm.cronLoop", func() { s.cronLoop(ctx) }) + helper.Go(s.log, "strm.queueCleanupLoop", func() { s.queueCleanupLoop(ctx) }) + helper.Go(s.log, "strm.refresh115TokensLoop", func() { s.refresh115TokensLoop(ctx) }) s.log.Info("strm service started", zap.Int("download_threads", downloadThreads), zap.Int("upload_threads", uploadThreads)) diff --git a/internal/service/strm_sync.go b/internal/service/strm_sync.go index 85fd59b..e8bcbcf 100644 --- a/internal/service/strm_sync.go +++ b/internal/service/strm_sync.go @@ -18,6 +18,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/service/cloud" "github.com/truewhile/MeBox/internal/service/cloud115" @@ -105,7 +106,7 @@ func (s *StrmService) StartSync(ctx context.Context, pathID string, syncType ... p.LastSyncMessage = "同步进行中" _ = s.repo.StrmSyncPath.Update(ctx, p) - go s.runSync(runCtx, p, rec) + helper.Go(s.log, "strm.sync", func() { s.runSync(runCtx, p, rec) }) return nil } @@ -361,38 +362,45 @@ func (st *strmSyncState) walkRemote() error { wg.Add(1) go func() { defer wg.Done() - for task := range queue { - if ctx.Err() != nil { - return - } - entries, err := st.provider.List(ctx, task.id) - if err != nil { - errMu.Lock() - if firstErr == nil { - firstErr = fmt.Errorf("列出远端目录 %s 失败:%w", task.id, err) + // worker 解析远端响应 panic 时取消整个同步,让 closer 与其余 + // worker 正常收尾,避免队列与 pending 计数卡死;正常退出不取消。 + if err := helper.Recover(st.s.log, "strm.sync.walkRemote", func() error { + for task := range queue { + if ctx.Err() != nil { + return nil } - errMu.Unlock() - cancel() - return - } - for _, entry := range entries { - cleanName := cleanEntryName(entry.Name, entry.IsDir) - rel := cleanName - if task.rel != "" { - rel = task.rel + "/" + cleanName - } - if entry.IsDir { - pending.Add(1) - select { - case queue <- dirTask{id: entry.ID, rel: rel}: - case <-ctx.Done(): - pending.Add(-1) + entries, err := st.provider.List(ctx, task.id) + if err != nil { + errMu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("列出远端目录 %s 失败:%w", task.id, err) } - } else { - st.processRemoteFile(entry, rel) + errMu.Unlock() + cancel() + return nil } + for _, entry := range entries { + cleanName := cleanEntryName(entry.Name, entry.IsDir) + rel := cleanName + if task.rel != "" { + rel = task.rel + "/" + cleanName + } + if entry.IsDir { + pending.Add(1) + select { + case queue <- dirTask{id: entry.ID, rel: rel}: + case <-ctx.Done(): + pending.Add(-1) + } + } else { + st.processRemoteFile(entry, rel) + } + } + pending.Add(-1) } - pending.Add(-1) + return nil + }); err != nil { + cancel() } }() } @@ -560,22 +568,28 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { wg.Add(1) go func() { defer wg.Done() - for t := range taskCh { - if ctx.Err() != nil { - return - } - files, _, err := open115.GetFsListFlat(ctx, rootCID, t.offset, pageSize) - if err != nil { - errMu.Lock() - if fetchErr == nil { - fetchErr = err + // 分页拉取 panic 时取消整个同步;正常退出不取消。 + if err := helper.Recover(st.s.log, "strm.sync.walk115.page", func() error { + for t := range taskCh { + if ctx.Err() != nil { + return nil } - errMu.Unlock() - return + files, _, err := open115.GetFsListFlat(ctx, rootCID, t.offset, pageSize) + if err != nil { + errMu.Lock() + if fetchErr == nil { + fetchErr = err + } + errMu.Unlock() + return nil + } + filesMu.Lock() + allFiles = append(allFiles, files...) + filesMu.Unlock() } - filesMu.Lock() - allFiles = append(allFiles, files...) - filesMu.Unlock() + return nil + }); err != nil { + cancel() } }() } @@ -632,63 +646,70 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { pwg.Add(1) go func() { defer pwg.Done() - for pid := range pidCh { - if ctx.Err() != nil { - return - } - if _, loaded := st.dirCache.Load(pid); loaded { - if n := doneDirs.Add(1); n%20 == 0 || n == int64(totalDirs) { + // 解析目录详情 panic 时中止整个同步(避免带着损坏的相对路径 + // 继续执行);正常退出不取消。 + if err := helper.Recover(st.s.log, "strm.sync.walk115.dirTree", func() error { + for pid := range pidCh { + if ctx.Err() != nil { + return nil + } + if _, loaded := st.dirCache.Load(pid); loaded { + if n := doneDirs.Add(1); n%20 == 0 || n == int64(totalDirs) { + st.updateSyncMessage(fmt.Sprintf("正在解析目录树 (%d/%d)...", n, totalDirs)) + } + continue + } + detail, err := open115.GetFsDetailByCid(ctx, pid) + if err != nil { + // 目录详情解析失败会导致下游文件 rel 无法还原真实父路径, + // seen key 与磁盘路径对不上:增量 prune 会误删本地文件、上传会 + // 误传本地未变文件、下载会重复下载。这里不是降级容错,而是 + // 直接中止整个同步——宁可本次同步失败,也不带着损坏的相对路径 + // 继续执行造成大规模误删/误传/重下(参考用户反馈"云盘没动却重下重传")。 + errMu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("115: 解析目录树失败(file_id=%s):%w", pid, err) + } + errMu.Unlock() + st.scanIncomplete.Store(true) + cancel() + return nil + } else if detail != nil { + // 解析相对路径 + relPath := cleanDirRel(detail.RelativePath(rootCID)) + st.dirCache.Store(pid, relPath) + _ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, pid, relPath) + + // 顺便解析并缓存 detail.Paths 中包含的中间各层级目录 + for _, ancestor := range detail.Paths { + if ancestor.FileId == "0" || ancestor.FileId == rootCID { + continue + } + if _, loaded := st.dirCache.Load(ancestor.FileId); !loaded { + subDetail := &cloud115.RemoteFileDetail{ + FileId: ancestor.FileId, + FileName: ancestor.Name, + Paths: nil, + } + for _, p := range detail.Paths { + subDetail.Paths = append(subDetail.Paths, p) + if p.FileId == ancestor.FileId { + break + } + } + ancestorRel := cleanDirRel(subDetail.RelativePath(rootCID)) + st.dirCache.Store(ancestor.FileId, ancestorRel) + _ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, ancestor.FileId, ancestorRel) + } + } + } + if n := doneDirs.Add(1); n%10 == 0 || n == int64(totalDirs) { st.updateSyncMessage(fmt.Sprintf("正在解析目录树 (%d/%d)...", n, totalDirs)) } - continue - } - detail, err := open115.GetFsDetailByCid(ctx, pid) - if err != nil { - // 目录详情解析失败会导致下游文件 rel 无法还原真实父路径, - // seen key 与磁盘路径对不上:增量 prune 会误删本地文件、上传会 - // 误传本地未变文件、下载会重复下载。这里不是降级容错,而是 - // 直接中止整个同步——宁可本次同步失败,也不带着损坏的相对路径 - // 继续执行造成大规模误删/误传/重下(参考用户反馈"云盘没动却重下重传")。 - errMu.Lock() - if firstErr == nil { - firstErr = fmt.Errorf("115: 解析目录树失败(file_id=%s):%w", pid, err) - } - errMu.Unlock() - st.scanIncomplete.Store(true) - cancel() - return - } else if detail != nil { - // 解析相对路径 - relPath := cleanDirRel(detail.RelativePath(rootCID)) - st.dirCache.Store(pid, relPath) - _ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, pid, relPath) - - // 顺便解析并缓存 detail.Paths 中包含的中间各层级目录 - for _, ancestor := range detail.Paths { - if ancestor.FileId == "0" || ancestor.FileId == rootCID { - continue - } - if _, loaded := st.dirCache.Load(ancestor.FileId); !loaded { - subDetail := &cloud115.RemoteFileDetail{ - FileId: ancestor.FileId, - FileName: ancestor.Name, - Paths: nil, - } - for _, p := range detail.Paths { - subDetail.Paths = append(subDetail.Paths, p) - if p.FileId == ancestor.FileId { - break - } - } - ancestorRel := cleanDirRel(subDetail.RelativePath(rootCID)) - st.dirCache.Store(ancestor.FileId, ancestorRel) - _ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, ancestor.FileId, ancestorRel) - } - } - } - if n := doneDirs.Add(1); n%10 == 0 || n == int64(totalDirs) { - st.updateSyncMessage(fmt.Sprintf("正在解析目录树 (%d/%d)...", n, totalDirs)) } + return nil + }); err != nil { + cancel() } }() } diff --git a/internal/service/subtitle.go b/internal/service/subtitle.go index 32b10ae..67a8831 100644 --- a/internal/service/subtitle.go +++ b/internal/service/subtitle.go @@ -23,6 +23,8 @@ import ( "path/filepath" "regexp" "strings" + "sync" + "time" "go.uber.org/zap" @@ -35,6 +37,21 @@ type SubtitleService struct { log *zap.Logger repo *repository.Container cfg *config.Config + + // 目录发现是 Emby 条目列表的热路径(每个媒体源一次 DB 查询 + 最多 5 次 + // os.ReadDir),而字幕文件极少变化:按 media_id 做短 TTL 缓存。 + cacheMu sync.Mutex + discovery map[string]subtitleDiscoveryEntry +} + +const ( + subtitleDiscoveryTTL = 2 * time.Minute + subtitleDiscoveryCacheCap = 4096 +) + +type subtitleDiscoveryEntry struct { + tracks []SubtitleTrack + expiresAt time.Time } // NewSubtitleService is the constructor. @@ -73,6 +90,50 @@ func (s *SubtitleService) DiscoverExternalOnly(ctx context.Context, mediaID stri } func (s *SubtitleService) discover(ctx context.Context, mediaID string) ([]SubtitleTrack, error) { + if tracks, ok := s.cachedDiscovery(mediaID); ok { + return tracks, nil + } + tracks, err := s.discoverUncached(ctx, mediaID) + if err != nil { + return nil, err + } + s.rememberDiscovery(mediaID, tracks) + return tracks, nil +} + +func (s *SubtitleService) cachedDiscovery(mediaID string) ([]SubtitleTrack, bool) { + now := time.Now() + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + entry, ok := s.discovery[mediaID] + if !ok { + return nil, false + } + if now.After(entry.expiresAt) { + delete(s.discovery, mediaID) + return nil, false + } + // 返回副本,避免调用方修改缓存内容。 + return append([]SubtitleTrack(nil), entry.tracks...), true +} + +func (s *SubtitleService) rememberDiscovery(mediaID string, tracks []SubtitleTrack) { + now := time.Now() + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + if s.discovery == nil { + s.discovery = make(map[string]subtitleDiscoveryEntry) + } + if len(s.discovery) >= subtitleDiscoveryCacheCap { + s.discovery = make(map[string]subtitleDiscoveryEntry) + } + s.discovery[mediaID] = subtitleDiscoveryEntry{ + tracks: append([]SubtitleTrack(nil), tracks...), + expiresAt: now.Add(subtitleDiscoveryTTL), + } +} + +func (s *SubtitleService) discoverUncached(ctx context.Context, mediaID string) ([]SubtitleTrack, error) { m, err := s.repo.Media.FindByID(ctx, mediaID) if err != nil { return nil, err diff --git a/internal/service/system_update.go b/internal/service/system_update.go index 36ceedd..2543e68 100644 --- a/internal/service/system_update.go +++ b/internal/service/system_update.go @@ -11,6 +11,7 @@ import ( "go.uber.org/zap" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/repository" ) @@ -132,7 +133,7 @@ func (s *SystemUpdateService) Apply(ctx context.Context) (SystemUpdateStatus, er s.last = &status s.mu.Unlock() - go s.runUpdate(context.Background(), status, task) + helper.Go(s.log, "system.update", func() { s.runUpdate(context.Background(), status, task) }) return status, nil } diff --git a/internal/service/token_svc_pending.go b/internal/service/token_svc_pending.go index c2eaf09..662bb20 100644 --- a/internal/service/token_svc_pending.go +++ b/internal/service/token_svc_pending.go @@ -7,6 +7,7 @@ import ( "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" ) @@ -26,11 +27,14 @@ func (s *TokenService) storeRefreshTokenBestEffort(userID, tokenHash string, exp 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, + err := helper.Recover(s.log, "token.storeRefreshToken", func() error { + return s.storeRefreshToken(ctx, &model.RefreshToken{ + UserID: userID, + TokenHash: tokenHash, + ExpiresAt: expiresAt, + }) }) + done <- err }() select { case err := <-done: diff --git a/internal/service/transcoder.go b/internal/service/transcoder.go index cfac991..780f85d 100644 --- a/internal/service/transcoder.go +++ b/internal/service/transcoder.go @@ -33,6 +33,7 @@ import ( "go.uber.org/zap" "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/repository" ) @@ -143,7 +144,7 @@ func (t *TranscoderService) EnsureJob(ctx context.Context, mediaID string) (stri t.jobs[mediaID] = job t.mu.Unlock() - go t.monitorIdle(jobCtx, job) - go t.runFFmpeg(jobCtx, job, m.Path) + helper.Go(t.log, "transcoder.monitorIdle", func() { t.monitorIdle(jobCtx, job) }) + helper.Go(t.log, "transcoder.ffmpeg", func() { t.runFFmpeg(jobCtx, job, m.Path) }) return t.PlaylistPath(mediaID), nil } diff --git a/internal/service/watcher.go b/internal/service/watcher.go index e72b14b..77958cf 100644 --- a/internal/service/watcher.go +++ b/internal/service/watcher.go @@ -22,6 +22,7 @@ import ( "github.com/fsnotify/fsnotify" "go.uber.org/zap" + "github.com/truewhile/MeBox/internal/helper" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" ) @@ -69,8 +70,8 @@ func (w *WatcherService) Start(ctx context.Context) error { if err := w.Refresh(ctx); err != nil { w.log.Warn("watcher refresh failed", zap.Error(err)) } - go w.loop(ctx) - go w.debouncer(ctx) + helper.Go(w.log, "watcher.loop", func() { w.loop(ctx) }) + helper.Go(w.log, "watcher.debouncer", func() { w.debouncer(ctx) }) return nil } diff --git a/web/index.html b/web/index.html index 55c74d7..505e9b5 100644 --- a/web/index.html +++ b/web/index.html @@ -8,12 +8,8 @@ - - - +
- ✓ 该账号已授权;重新授权会替换现有令牌 -
- )} - - {!authUI && ( -未找到匹配的官方应用
} -{authStatus || '等待扫码…'}
-请使用 115 手机客户端扫码并确认授权,5 分钟内有效
-请点击下方按钮在新窗口完成授权:
- -完成后回到此页面等待自动确认(授权成功后可关闭弹窗)
-+ ✓ 该账号已授权;重新授权会替换现有令牌 +
+ )} + + {!authUI && ( +未找到匹配的官方应用
} +{authStatus || '等待扫码…'}
+请使用 115 手机客户端扫码并确认授权,5 分钟内有效
+请点击下方按钮在新窗口完成授权:
+ +完成后回到此页面等待自动确认(授权成功后可关闭弹窗)
+