From 984b98e61b3b92495683a0505a6800dbc5acad97 Mon Sep 17 00:00:00 2001 From: jason jia Date: Sat, 17 Oct 2026 01:29:51 +0000 Subject: [PATCH 1/2] fix: preserve database compatibility and harden request lifecycles --- .github/workflows/ci.yml | 35 ++- CONTRIBUTING.md | 9 + cmd/server/server_manager.go | 3 + internal/database/compatibility_test.go | 238 ++++++++++++++++++ internal/database/persisted_rows.go | 50 ++++ internal/database/schema_migration.go | 24 +- internal/database/sqlite_migration_copy.go | 2 +- internal/database/sqlite_runtime.go | 30 +-- internal/database/sqlite_runtime_test.go | 79 ++++++ internal/handler/media_test.go | 4 +- internal/repository/media_repository.go | 22 +- .../repository/media_repository_upsert.go | 23 +- .../repository/media_search_repository.go | 23 +- internal/service/backup_postgres.go | 3 +- internal/service/backup_zero_values_test.go | 34 +++ internal/service/ffmpeg_tools_test.go | 2 + internal/service/image_proxy.go | 64 +---- .../image_proxy_dial_allowlist_test.go | 36 ++- internal/service/image_proxy_remote_fetch.go | 4 +- internal/service/image_proxy_transport.go | 67 +++++ .../service/transcoder_availability_test.go | 2 + web/src/pages/reader/ReaderReplacePage.tsx | 1 - web/src/pages/reader/ReaderSourcesPage.tsx | 1 - web/src/pages/reader/ReaderViewPage.tsx | 133 +++++----- 24 files changed, 687 insertions(+), 202 deletions(-) create mode 100644 internal/database/compatibility_test.go create mode 100644 internal/database/persisted_rows.go create mode 100644 internal/database/sqlite_runtime_test.go create mode 100644 internal/service/backup_zero_values_test.go create mode 100644 internal/service/image_proxy_transport.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e416885..d9da205 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -40,6 +40,39 @@ jobs: - name: go test run: go test ./... + database-compatibility: + name: SQLite / PostgreSQL ${{ matrix.postgres }} + runs-on: ubuntu-latest + strategy: + matrix: + postgres: [16, 17] + services: + postgres: + image: postgres:${{ matrix.postgres }}-alpine + env: + POSTGRES_USER: mebox + POSTGRES_PASSWORD: mebox-test-only + POSTGRES_DB: mebox_test + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U mebox -d mebox_test" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + env: + MEBOX_TEST_POSTGRES_DSN: postgres://mebox:mebox-test-only@127.0.0.1:5432/mebox_test?sslmode=disable + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version: '1.26' + cache: true + + - name: Database regression tests + run: go test ./internal/database ./internal/repository -count=1 + frontend: name: Frontend (Node) runs-on: ubuntu-latest @@ -78,7 +111,7 @@ jobs: ci-success: name: CI Success runs-on: ubuntu-latest - needs: [backend, frontend, smoke] + needs: [backend, frontend, database-compatibility, smoke] if: success() steps: - run: echo "CI passed, ready for release" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 22fccd6..ed4fad8 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -84,6 +84,15 @@ go test ./internal/service -run "TestOrganize|TestSubscription|TestMTeam" -count go test ./cmd/server -count=1 ``` +数据库兼容测试默认覆盖 SQLite。连接专用测试 PostgreSQL 后可运行真实的双数据库回归： + +```bash +MEBOX_TEST_POSTGRES_DSN='postgres://mebox:test-password@127.0.0.1:5432/mebox_test?sslmode=disable' \ + go test ./internal/database ./internal/repository -count=1 +``` + +测试账号需要创建、删除 schema 的权限；每个用例使用独立的临时 schema。CI 会分别在 PostgreSQL 16、17 上执行这些检查。 + ### 代码要求 - Go 代码使用 `gofmt`。 diff --git a/cmd/server/server_manager.go b/cmd/server/server_manager.go index 2568521..ccd4eff 100644 --- a/cmd/server/server_manager.go +++ b/cmd/server/server_manager.go @@ -355,6 +355,9 @@ func (m *serverManager) drain(s *http.Server) { if s == nil { return } + // Stop reusing old connections before Reload returns. Asynchronous Shutdown + // alone leaves a window where a client receives the previous HTTP/TLS mode. + s.SetKeepAlivesEnabled(false) go func(s *http.Server) { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() diff --git a/internal/database/compatibility_test.go b/internal/database/compatibility_test.go new file mode 100644 index 0000000..7676786 --- /dev/null +++ b/internal/database/compatibility_test.go @@ -0,0 +1,238 @@ +package database + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/stdlib" + "gorm.io/driver/postgres" + "gorm.io/gorm" + "gorm.io/gorm/logger" + + "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/repository" +) + +// PostgreSQL tests use a disposable schema, never the caller's existing tables. +// Set MEBOX_TEST_POSTGRES_DSN to include a real PostgreSQL server in local runs. +func compatibilityDB(t *testing.T, dialect string) *gorm.DB { + t.Helper() + var db *gorm.DB + if dialect == "sqlite" { + cfg := &config.Config{} + cfg.Database.Type = dialect + cfg.Database.DBPath = filepath.Join(t.TempDir(), "test.db") + cfg.Database.WALMode = true + cfg.Database.BusyTimeout = 5000 + cfg.Database.MaxOpenConns = 8 + var err error + db, err = Open(cfg, nil) + if err != nil { + t.Fatal(err) + } + } else { + dsn := os.Getenv("MEBOX_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set MEBOX_TEST_POSTGRES_DSN to run PostgreSQL integration tests") + } + pgcfg, err := pgx.ParseConfig(dsn) + if err != nil { + t.Fatal("invalid PostgreSQL test configuration") + } + admin := stdlib.OpenDB(*pgcfg) + t.Cleanup(func() { _ = admin.Close() }) + schemaName := "mebox_test_" + strings.ReplaceAll(uuid.NewString(), "-", "") + if _, err := admin.Exec("CREATE SCHEMA " + quoteIdent(schemaName)); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if _, err := admin.Exec("DROP SCHEMA " + quoteIdent(schemaName) + " CASCADE"); err != nil { + t.Error(err) + } + }) + pgcfg.RuntimeParams["search_path"] = schemaName + ",public" + conn := stdlib.OpenDB(*pgcfg) + t.Cleanup(func() { _ = conn.Close() }) + db, err = gorm.Open(postgres.New(postgres.Config{Conn: conn}), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + } + db.Logger = logger.Default.LogMode(logger.Silent) + if conn, err := db.DB(); err == nil { + t.Cleanup(func() { _ = conn.Close() }) + } + if err := AutoMigrate(db); err != nil { + t.Fatal(err) + } + return db +} + +func TestDatabaseCompatibility(t *testing.T) { + for _, dialect := range []string{"sqlite", "postgres"} { + t.Run(dialect, func(t *testing.T) { + t.Run("long_paths_and_restart", func(t *testing.T) { + db := compatibilityDB(t, dialect) + // Distinct CJK characters prevent TOAST compression from hiding + // the PostgreSQL btree entry size limit (> 2704 bytes). + var path strings.Builder + path.WriteString("/media/") + for i := range 980 { + path.WriteRune(rune(0x4e00 + i)) + } + path.WriteString(".mkv") + media := model.Media{Path: path.String(), Title: "Long path"} + if err := db.Create(&media).Error; err != nil { + t.Fatalf("insert long media path: %v", err) + } + if err := db.Create(&model.Media{Path: media.Path}).Error; err == nil { + t.Fatal("duplicate path accepted") + } + if err := AutoMigrate(db); err != nil { + t.Fatalf("migrate existing database: %v", err) + } + var found model.Media + if err := db.Where("path = ?", media.Path).First(&found).Error; err != nil { + t.Fatalf("lookup long path: %v", err) + } + }) + if dialect == "postgres" { + t.Run("upgrade_legacy_path_indexes", func(t *testing.T) { + db := compatibilityDB(t, dialect) + for _, unique := range []string{"UNIQUE ", ""} { + if err := db.Exec("DROP INDEX idx_media_path").Error; err != nil { + t.Fatal(err) + } + if err := db.Exec("CREATE " + unique + "INDEX idx_media_path ON media(path)").Error; err != nil { + t.Fatal(err) + } + for range 2 { + if err := ensureMediaPathIndex(db); err != nil { + t.Fatal(err) + } + } + isUnique, method, found, err := postgresIndexProperties(db, "idx_media_path") + if err != nil || !found || isUnique || method != "hash" { + t.Fatalf("legacy index not upgraded: unique=%v method=%s found=%v err=%v", isUnique, method, found, err) + } + } + }) + } + t.Run("search_ranking", func(t *testing.T) { + db := compatibilityDB(t, dialect) + // Exercise LIKE fallback in both engines. + if dialect == "sqlite" { + resetMediaSearchIndex(db) + } + for i, title := range []string{"The Alpha story", "Alpha sequel", "Alpha"} { + row := model.Media{Base: model.Base{ID: fmt.Sprintf("m-%d", i), CreatedAt: time.Now().Add(time.Duration(i) * time.Minute)}, Title: title, Path: fmt.Sprintf("/media/%d.mkv", i)} + if err := db.Create(&row).Error; err != nil { + t.Fatal(err) + } + } + items, total, err := repository.New(db).Media.SearchFilteredPage(t.Context(), "Alpha", 0, 10, repository.MediaQueryFilter{IncludeNSFW: true}) + if err != nil { + t.Fatal(err) + } + if total != 3 || len(items) != 3 || items[0].Title != "Alpha" || items[1].Title != "Alpha sequel" { + t.Fatalf("unexpected search ranking: total=%d, items=%v", total, items) + } + }) + t.Run("concurrent_media_upsert", func(t *testing.T) { + db := compatibilityDB(t, dialect) + repo := repository.New(db).Media + const workers = 8 + // Synchronize the initial lookups to force the insert race. + var lookups sync.WaitGroup + lookups.Add(workers) + if dialect == "postgres" { + if err := db.Callback().Query().After("gorm:after_query").Register("test:insert_race", func(tx *gorm.DB) { + if tx.Statement.Table == "media" && tx.Error == gorm.ErrRecordNotFound { + lookups.Done() + lookups.Wait() + } + }); err != nil { + t.Fatal(err) + } + } + ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) + defer cancel() + errs := make(chan error, workers) + for range workers { + go func() { errs <- repo.Upsert(ctx, &model.Media{Path: "/media/shared.mkv", Title: "Shared"}) }() + } + for range workers { + if err := <-errs; err != nil { + t.Errorf("concurrent upsert: %v", err) + } + } + var count int64 + if err := db.Model(&model.Media{}).Count(&count).Error; err != nil || count != 1 { + t.Fatalf("media count=%d, err=%v", count, err) + } + }) + t.Run("search_filters", func(t *testing.T) { + db := compatibilityDB(t, dialect) + rows := []model.Media{ + {Base: model.Base{ID: "keep"}, LibraryID: "lib-1", Genres: "Action", Year: 2020, Rating: 8}, + {Base: model.Base{ID: "other-library"}, LibraryID: "lib-2", Genres: "Action", Year: 2020, Rating: 8}, + {Base: model.Base{ID: "other-genre"}, LibraryID: "lib-1", Genres: "Comedy", Year: 2020, Rating: 8}, + {Base: model.Base{ID: "too-old"}, LibraryID: "lib-1", Genres: "Action", Year: 1990, Rating: 8}, + {Base: model.Base{ID: "too-new"}, LibraryID: "lib-1", Genres: "Action", Year: 2026, Rating: 8}, + {Base: model.Base{ID: "low-rating"}, LibraryID: "lib-1", Genres: "Action", Year: 2020, Rating: 2}, + {Base: model.Base{ID: "watched"}, LibraryID: "lib-1", Genres: "Action", Year: 2020, Rating: 8}, + } + for i := range rows { + rows[i].Title = "Alpha " + rows[i].ID + rows[i].Path = "/media/" + rows[i].ID + ".mkv" + if err := db.Create(&rows[i]).Error; err != nil { + t.Fatal(err) + } + } + if err := db.Create(&model.PlaybackHistory{UserID: "viewer", MediaID: "watched", Completed: true}).Error; err != nil { + t.Fatal(err) + } + items, total, err := repository.New(db).Media.SearchFilteredPage(t.Context(), "alpha", 0, 10, repository.MediaQueryFilter{ + IncludeNSFW: true, LibraryID: "lib-1", Genres: []string{"Action"}, YearMin: 2000, YearMax: 2025, RatingMin: 5, + UnwatchedOnly: true, UnwatchedUserID: "viewer", + }) + if err != nil { + t.Fatal(err) + } + if total != 1 || len(items) != 1 || items[0].ID != "keep" { + t.Fatalf("search ignored filters: total=%d, items=%v", total, items) + } + }) + t.Run("migration_preserves_zero_values", func(t *testing.T) { + src := compatibilityDB(t, "sqlite") + dst := compatibilityDB(t, dialect) + user := model.User{Username: "disabled", PasswordHash: "hash"} + if err := src.Create(&user).Error; err != nil { + t.Fatal(err) + } + if err := src.Model(&user).Updates(map[string]any{"is_active": false, "hide_adult": false, "player_volume": 0, "danmaku_enabled": false}).Error; err != nil { + t.Fatal(err) + } + if _, _, err := copyModelTables(src, dst, 500); err != nil { + t.Fatal(err) + } + var got model.User + if err := dst.First(&got, "id = ?", user.ID).Error; err != nil { + t.Fatal(err) + } + if got.IsActive || got.HideAdult || got.PlayerVolume != 0 || got.DanmakuEnabled { + t.Fatalf("migration replaced stored zero values: active=%v hide_adult=%v volume=%v danmaku=%v", got.IsActive, got.HideAdult, got.PlayerVolume, got.DanmakuEnabled) + } + }) + }) + } +} diff --git a/internal/database/persisted_rows.go b/internal/database/persisted_rows.go new file mode 100644 index 0000000..f565992 --- /dev/null +++ b/internal/database/persisted_rows.go @@ -0,0 +1,50 @@ +package database + +import ( + "fmt" + "reflect" + + "gorm.io/gorm" +) + +// InsertPreservingValues copies persisted model rows without applying creation +// hooks or defaults. Struct CreateInBatches treats false, 0 and empty strings as +// missing values, which would reactivate disabled accounts during migration or +// restore. Column maps preserve the exact values already stored in the source. +func InsertPreservingValues(db *gorm.DB, rows any, batchSize int) error { + stmt := &gorm.Statement{DB: db} + if err := stmt.Parse(rows); err != nil { + return err + } + slice := reflect.Indirect(reflect.ValueOf(rows)) + if slice.Kind() != reflect.Slice { + return fmt.Errorf("persisted rows must be a slice, got %T", rows) + } + if len(stmt.Schema.DBNames) == 0 { + return fmt.Errorf("persisted rows have no database columns") + } + values := make([]map[string]any, 0, slice.Len()) + for i := 0; i < slice.Len(); i++ { + row := make(map[string]any, len(stmt.Schema.DBNames)) + for _, name := range stmt.Schema.DBNames { + field := stmt.Schema.FieldsByDBName[name] + if field.Creatable { + row[name], _ = field.ValueOf(db.Statement.Context, slice.Index(i)) + } + } + values = append(values, row) + } + if len(values) == 0 { + return nil + } + // Bound batches by each engine's bind-parameter limit. + parameterLimit := 65535 + if isSQLite(db) { + parameterLimit = 32766 + } + maxBatch := parameterLimit / len(stmt.Schema.DBNames) + if batchSize <= 0 || batchSize > maxBatch { + batchSize = maxBatch + } + return db.Table(stmt.Schema.Table).CreateInBatches(values, batchSize).Error +} diff --git a/internal/database/schema_migration.go b/internal/database/schema_migration.go index 6839c32..1313702 100644 --- a/internal/database/schema_migration.go +++ b/internal/database/schema_migration.go @@ -71,8 +71,8 @@ func AutoMigrateWithConfig(db *gorm.DB, cfg *config.Config, log *zap.Logger) err // index row size ... exceeds btree version 4 maximum 2704 // // 所以 PostgreSQL 改用 md5(path) 的表达式唯一索引（定长 32 字节，永远塞得下）， -// 同时保留一条普通索引 idx_media_path 供 path = ? 等值查找（扫描与 upsert 的 -// 热路径）使用。SQLite 没有这个长度限制，沿用整列唯一索引。 +// 同时用 hash 索引 idx_media_path 加速 path = ? 等值查找；非唯一 btree +// 也有索引项长度限制，不能继续对完整 path 建 btree。SQLite 沿用整列唯一索引。 func ensureMediaPathIndex(db *gorm.DB) error { if db == nil || !db.Migrator().HasTable("media") { return nil @@ -84,37 +84,39 @@ func ensureMediaPathIndex(db *gorm.DB) error { if err := db.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_media_path_hash ON media(md5(path))`).Error; err != nil { return err } - unique, found, err := postgresIndexIsUnique(db, "idx_media_path") + unique, method, found, err := postgresIndexProperties(db, "idx_media_path") if err != nil { return err } - if found && unique { + if found && (unique || method != "hash") { if err := db.Exec(`DROP INDEX IF EXISTS idx_media_path`).Error; err != nil { return err } } - return db.Exec(`CREATE INDEX IF NOT EXISTS idx_media_path ON media(path)`).Error + return db.Exec(`CREATE INDEX IF NOT EXISTS idx_media_path ON media USING hash(path)`).Error } -// postgresIndexIsUnique 报告当前 schema 下是否存在该索引、以及它是否唯一。 -func postgresIndexIsUnique(db *gorm.DB, indexName string) (unique bool, found bool, err error) { +// postgresIndexProperties 返回当前 schema 下索引的唯一性与访问方法。 +func postgresIndexProperties(db *gorm.DB, indexName string) (unique bool, method string, found bool, err error) { // 用指针接 bool：没有匹配行时 Scan 不报错，nil 即代表索引不存在。 var row struct { IsUnique *bool + Method string } if err := db.Raw(` -SELECT i.indisunique AS is_unique +SELECT i.indisunique AS is_unique, a.amname AS method FROM pg_class c JOIN pg_index i ON i.indexrelid = c.oid JOIN pg_namespace n ON n.oid = c.relnamespace +JOIN pg_am a ON a.oid = c.relam WHERE n.nspname = current_schema() AND c.relname = ? `, indexName).Scan(&row).Error; err != nil { - return false, false, err + return false, "", false, err } if row.IsUnique == nil { - return false, false, nil + return false, "", false, nil } - return *row.IsUnique, true, nil + return *row.IsUnique, row.Method, true, nil } // ensurePostgresSearchIndexes 给 PostgreSQL 补上 SQLite FTS5 的替代品。 diff --git a/internal/database/sqlite_migration_copy.go b/internal/database/sqlite_migration_copy.go index 038ec20..c727d35 100644 --- a/internal/database/sqlite_migration_copy.go +++ b/internal/database/sqlite_migration_copy.go @@ -78,7 +78,7 @@ func copyModelTables(src, target *gorm.DB, batchSize int) (map[string]int64, int if filtered.Len() > 0 { filteredPtr := reflect.New(filtered.Type()) filteredPtr.Elem().Set(filtered) - if err := target.Clauses(clause.OnConflict{DoNothing: true}).CreateInBatches(filteredPtr.Interface(), batchSize).Error; err != nil { + if err := InsertPreservingValues(target.Clauses(clause.OnConflict{DoNothing: true}), filteredPtr.Interface(), batchSize); err != nil { return tableCounts, totalCopied, fmt.Errorf("copy sqlite table %s: %w", table, err) } copiedForTable += int64(filtered.Len()) diff --git a/internal/database/sqlite_runtime.go b/internal/database/sqlite_runtime.go index f06d3df..2129ee7 100644 --- a/internal/database/sqlite_runtime.go +++ b/internal/database/sqlite_runtime.go @@ -104,19 +104,18 @@ func (g *sqliteWriteGate) Lock(ctx context.Context) (*sqliteGateHolder, error) { if ctx == nil { ctx = context.Background() } - // ctx 取消时唤醒等待者（cond 无法感知 ctx，用旁路 goroutine 广播）。 - if done := ctx.Done(); done != nil { - stop := make(chan struct{}) - defer close(stop) - go func() { - select { - case <-done: - g.cond.Broadcast() - case <-stop: - } - }() + // 广播也必须持有同一把锁，否则取消可能落在检查与 Wait 之间而丢失。 + wake := func() { + g.mu.Lock() + defer g.mu.Unlock() + g.cond.Broadcast() } + stopCancel := context.AfterFunc(ctx, wake) + defer stopCancel() for { + if ctx.Err() != nil { + return nil, ctx.Err() + } if g.owner == nil { holder := &sqliteGateHolder{ id: sqliteGateHolderSeq.Add(1), @@ -125,16 +124,17 @@ func (g *sqliteWriteGate) Lock(ctx context.Context) (*sqliteGateHolder, error) { g.owner = holder return holder, nil } - if ctx.Err() != nil { - return nil, ctx.Err() - } - if time.Since(g.owner.acquired) > sqliteGateHoldLimit { + remaining := sqliteGateHoldLimit - time.Since(g.owner.acquired) + if remaining <= 0 { // 持有者疑似 panic 泄漏（After 回调未执行）：强制回收。 g.owner = nil g.cond.Broadcast() continue } + // 无后续写入或取消时，也要在租约到期后主动唤醒等待者。 + timer := time.AfterFunc(remaining, wake) g.cond.Wait() + timer.Stop() } } diff --git a/internal/database/sqlite_runtime_test.go b/internal/database/sqlite_runtime_test.go new file mode 100644 index 0000000..e6346e7 --- /dev/null +++ b/internal/database/sqlite_runtime_test.go @@ -0,0 +1,79 @@ +package database + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestSQLiteWriteGateRejectsCancelledContext(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + gate := newSQLiteWriteGate() + holder, err := gate.Lock(ctx) + defer gate.Unlock(holder) + if !errors.Is(err, context.Canceled) { + t.Fatalf("Lock error=%v, want context.Canceled", err) + } +} + +func TestSQLiteWriteGateWakesForExpiredHolder(t *testing.T) { + gate := newSQLiteWriteGate() + holder, err := gate.Lock(t.Context()) + if err != nil { + t.Fatal(err) + } + defer gate.Unlock(holder) + // Make the lease expire shortly, without waiting a full minute. + gate.mu.Lock() + holder.acquired = time.Now().Add(-sqliteGateHoldLimit + 20*time.Millisecond) + gate.mu.Unlock() + done := make(chan *sqliteGateHolder, 1) + go func() { + next, _ := gate.Lock(t.Context()) + done <- next + }() + select { + case next := <-done: + gate.Unlock(holder) // A stale holder must not unlock its successor. + gate.mu.Lock() + owner := gate.owner + gate.mu.Unlock() + gate.Unlock(next) + if next == nil || owner != next { + t.Fatal("expired holder released the new owner's lock") + } + case <-time.After(time.Second): + gate.Unlock(holder) + gate.Unlock(<-done) + t.Fatal("waiter was not woken when the holder expired") + } +} + +func TestSQLiteWriteGateWakesForCancellation(t *testing.T) { + gate := newSQLiteWriteGate() + holder, err := gate.Lock(t.Context()) + if err != nil { + t.Fatal(err) + } + defer gate.Unlock(holder) + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { + next, err := gate.Lock(ctx) + gate.Unlock(next) + done <- err + }() + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("Lock error=%v, want context.Canceled", err) + } + case <-time.After(time.Second): + gate.Unlock(holder) + <-done + t.Fatal("waiter was not woken for cancellation") + } +} diff --git a/internal/handler/media_test.go b/internal/handler/media_test.go index 97423d3..a0e4aca 100644 --- a/internal/handler/media_test.go +++ b/internal/handler/media_test.go @@ -26,7 +26,9 @@ import ( ) func TestQueueLibraryRootScanImportsNewLibraryMedia(t *testing.T) { - db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + // The scanner and polling loop use separate connections; :memory: gives + // each connection its own empty database. + db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "scan.db")+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)"), &gorm.Config{}) if err != nil { t.Fatal(err) } diff --git a/internal/repository/media_repository.go b/internal/repository/media_repository.go index 673bf86..091844f 100644 --- a/internal/repository/media_repository.go +++ b/internal/repository/media_repository.go @@ -63,38 +63,38 @@ type MediaQueryFilter struct { func applyMediaQueryFilter(q *gorm.DB, filter MediaQueryFilter) *gorm.DB { if !filter.IncludeNSFW { - q = q.Where("nsfw = ?", false) + q = q.Where("media.nsfw = ?", false) } if len(filter.HiddenLibraryIDs) > 0 { - q = q.Where("library_id NOT IN ?", filter.HiddenLibraryIDs) + q = q.Where("media.library_id NOT IN ?", filter.HiddenLibraryIDs) } if len(filter.AllowedLibraryIDs) > 0 { - q = q.Where("library_id IN ?", filter.AllowedLibraryIDs) + q = q.Where("media.library_id IN ?", filter.AllowedLibraryIDs) } if libraryID := strings.TrimSpace(filter.LibraryID); libraryID != "" { - q = q.Where("library_id = ?", libraryID) + q = q.Where("media.library_id = ?", libraryID) } if seriesID := strings.TrimSpace(filter.SeriesID); seriesID != "" { - q = q.Where("series_id = ?", seriesID) + q = q.Where("media.series_id = ?", seriesID) } if len(filter.Genres) > 0 { q = q.Where(genreMatchClause(filter.Genres), genreMatchArgs(filter.Genres)...) } if filter.YearMin > 0 { - q = q.Where("year >= ?", filter.YearMin) + q = q.Where("media.year >= ?", filter.YearMin) } if filter.YearMax > 0 { - q = q.Where("year <= ?", filter.YearMax) + q = q.Where("media.year <= ?", filter.YearMax) } if filter.RatingMin > 0 { - q = q.Where("rating >= ?", filter.RatingMin) + q = q.Where("media.rating >= ?", filter.RatingMin) } if filter.UnwatchedOnly { userID := strings.TrimSpace(filter.UnwatchedUserID) // 没有用户上下文时忽略该条件：否则会把整个库筛成空，看起来像「坏了」。 if userID != "" { q = q.Where( - "id NOT IN (SELECT media_id FROM playback_histories WHERE user_id = ? AND completed = ?)", + "media.id NOT IN (SELECT media_id FROM playback_histories WHERE user_id = ? AND completed = ?)", userID, true, ) } @@ -113,7 +113,7 @@ func applyMediaQueryFilter(q *gorm.DB, filter MediaQueryFilter) *gorm.DB { func genreMatchClause(genres []string) string { clauses := make([]string, 0, len(genres)) for range genres { - clauses = append(clauses, "',' || REPLACE(REPLACE(TRIM(genres), ' ', ''), '，', ',') || ',' LIKE ?") + clauses = append(clauses, "',' || REPLACE(REPLACE(TRIM(media.genres), ' ', ''), '，', ',') || ',' LIKE ?") } return "(" + strings.Join(clauses, " OR ") + ")" } @@ -122,7 +122,7 @@ func genreMatchClause(genres []string) string { // // 必须与 genreMatchClause 的列端处理完全对称： // - TRIM → TrimSpace -// - REPLACE(…, ' ', '') → ReplaceAll(…, " ", "") ← 多词类型（如 "Science Fiction"） +// - REPLACE(…, ' ', ”) → ReplaceAll(…, " ", "") ← 多词类型（如 "Science Fiction"） // - REPLACE(…, '，', ',') → ReplaceAll(…, "，", ",") func genreMatchArgs(genres []string) []any { args := make([]any, 0, len(genres)) diff --git a/internal/repository/media_repository_upsert.go b/internal/repository/media_repository_upsert.go index 4a07e03..84b0554 100644 --- a/internal/repository/media_repository_upsert.go +++ b/internal/repository/media_repository_upsert.go @@ -7,6 +7,7 @@ import ( "strings" "gorm.io/gorm" + "gorm.io/gorm/clause" "github.com/truewhile/MeBox/internal/model" ) @@ -95,6 +96,9 @@ func (r *MediaRepository) UpsertBatchWithAliases(ctx context.Context, items []Me // indexByIDBestEffort 在事务提交后按 ID 回读最新行并同步搜索索引。 func (r *MediaRepository) indexByIDBestEffort(ctx context.Context, ids []string) { + if _, ok := r.searchBackend.(MediaSearchSyncBackend); !ok { + return + } for _, id := range ids { if id == "" { continue @@ -193,14 +197,19 @@ func (r *MediaRepository) findOrCreateMediaByPath(ctx context.Context, db *gorm. if m.ScrapeStatus == "" { m.ScrapeStatus = "pending" } - if createErr := db.WithContext(ctx).Create(m).Error; createErr == nil { - return *m, true, false, nil - } else if retryErr := db.WithContext(ctx).Unscoped().Where("path = ?", m.Path).First(&existing).Error; retryErr != nil { - return model.Media{}, false, false, createErr - } else { - // 并发插入竞态：重查已命中既有行，直接走更新分支。 - return existing, false, false, nil + // PostgreSQL 唯一键异常会中止整个事务，之后重查也会失败。 + // DO NOTHING 保持事务可用，并兼容 path 的表达式唯一索引。 + result := db.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}).Create(m) + if result.Error != nil { + return model.Media{}, false, false, result.Error } + if result.RowsAffected > 0 { + return *m, true, false, nil + } + if err := db.WithContext(ctx).Unscoped().Where("path = ?", m.Path).First(&existing).Error; err != nil { + return model.Media{}, false, false, err + } + return existing, false, false, nil } if err != nil { return model.Media{}, false, false, err diff --git a/internal/repository/media_search_repository.go b/internal/repository/media_search_repository.go index 63fed41..b3e54ce 100644 --- a/internal/repository/media_search_repository.go +++ b/internal/repository/media_search_repository.go @@ -6,6 +6,7 @@ import ( "unicode" "gorm.io/gorm" + "gorm.io/gorm/clause" "github.com/truewhile/MeBox/internal/model" ) @@ -86,7 +87,7 @@ func (r *MediaRepository) searchFilteredFTS(ctx context.Context, query string, o 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) + q = applyMediaQueryFilter(q, filter) if err := q.Count(&total).Error; err != nil { return nil, 0, false } @@ -122,10 +123,10 @@ func (r *MediaRepository) searchFilteredLIKE(ctx context.Context, query string, if query != "" { prefix := escapeLike(query) + "%" exact := query - q = q.Order(gorm.Expr( + q = q.Order(clause.OrderBy{Expression: gorm.Expr( "CASE WHEN title = ? THEN 0 WHEN original_name = ? THEN 1 WHEN title "+likeOp+" ? ESCAPE '\\' THEN 2 WHEN original_name "+likeOp+" ? ESCAPE '\\' THEN 3 ELSE 4 END, created_at desc", exact, exact, prefix, prefix, - )) + )}) } else { q = q.Order("created_at desc") } @@ -133,22 +134,6 @@ func (r *MediaRepository) searchFilteredLIKE(ctx context.Context, query string, return items, total, err } -func applyQualifiedMediaQueryFilter(q *gorm.DB, filter MediaQueryFilter) *gorm.DB { - if !filter.IncludeNSFW { - q = q.Where("media.nsfw = ?", false) - } - if len(filter.HiddenLibraryIDs) > 0 { - q = q.Where("media.library_id NOT IN ?", filter.HiddenLibraryIDs) - } - if len(filter.AllowedLibraryIDs) > 0 { - q = q.Where("media.library_id IN ?", filter.AllowedLibraryIDs) - } - if seriesID := strings.TrimSpace(filter.SeriesID); seriesID != "" { - q = q.Where("media.series_id = ?", seriesID) - } - return q -} - func mediaFTSQuery(query string) string { terms := mediaSearchTerms(query) if len(terms) == 0 { diff --git a/internal/service/backup_postgres.go b/internal/service/backup_postgres.go index 0dd6db7..1a754bc 100644 --- a/internal/service/backup_postgres.go +++ b/internal/service/backup_postgres.go @@ -15,6 +15,7 @@ import ( "gorm.io/gorm" "gorm.io/gorm/schema" + "github.com/truewhile/MeBox/internal/database" "github.com/truewhile/MeBox/internal/model" ) @@ -271,7 +272,7 @@ func (b *BackupService) restorePostgresBackup(ctx context.Context, src string) e if err != nil { return fmt.Errorf("表 %s 解码失败: %w", chunk.Table, err) } - if err := tx.Unscoped().CreateInBatches(rows, backupRestoreBatchSize).Error; err != nil { + if err := database.InsertPreservingValues(tx.Unscoped(), rows, backupRestoreBatchSize); err != nil { return fmt.Errorf("恢复表 %s 失败: %w", chunk.Table, err) } } diff --git a/internal/service/backup_zero_values_test.go b/internal/service/backup_zero_values_test.go new file mode 100644 index 0000000..4591a52 --- /dev/null +++ b/internal/service/backup_zero_values_test.go @@ -0,0 +1,34 @@ +package service + +import ( + "testing" + + "github.com/truewhile/MeBox/internal/model" +) + +func TestPostgresLogicalBackupPreservesZeroValues(t *testing.T) { + svc, _ := newBackupTestService(t) + user := model.User{Username: "disabled", PasswordHash: "hash"} + if err := svc.db.Create(&user).Error; err != nil { + t.Fatal(err) + } + if err := svc.db.Model(&user).Updates(map[string]any{ + "is_active": false, "hide_adult": false, "player_volume": 0, "danmaku_enabled": false, + }).Error; err != nil { + t.Fatal(err) + } + dump, err := svc.createPostgresBackup(t.Context(), "zero-values") + if err != nil { + t.Fatal(err) + } + if err := svc.restorePostgresBackup(t.Context(), dump); err != nil { + t.Fatal(err) + } + var got model.User + if err := svc.db.First(&got, "id = ?", user.ID).Error; err != nil { + t.Fatal(err) + } + if got.IsActive || got.HideAdult || got.PlayerVolume != 0 || got.DanmakuEnabled { + t.Fatalf("restore replaced zero values: active=%v hide_adult=%v volume=%v danmaku=%v", got.IsActive, got.HideAdult, got.PlayerVolume, got.DanmakuEnabled) + } +} diff --git a/internal/service/ffmpeg_tools_test.go b/internal/service/ffmpeg_tools_test.go index 77db7e3..a4cc263 100644 --- a/internal/service/ffmpeg_tools_test.go +++ b/internal/service/ffmpeg_tools_test.go @@ -49,6 +49,8 @@ func TestSafeZipTargetRejectsTraversal(t *testing.T) { } func TestFFmpegToolsStatusNoPanic(t *testing.T) { + t.Setenv("PATH", t.TempDir()) + t.Chdir(t.TempDir()) svc := NewFFmpegToolsService(&config.Config{}, zap.NewNop(), nil) st := svc.Status(context.Background()) for _, key := range []string{"installing", "message", "error", "install_dir", "target", "ffmpeg", "ffprobe"} { diff --git a/internal/service/image_proxy.go b/internal/service/image_proxy.go index 5307b44..598042f 100644 --- a/internal/service/image_proxy.go +++ b/internal/service/image_proxy.go @@ -14,14 +14,12 @@ package service import ( "context" - "errors" "net" "net/http" "path/filepath" "runtime" "strings" "sync" - "syscall" "time" "go.uber.org/zap" @@ -103,68 +101,14 @@ func NewImageProxy(cfg *config.Config, log *zap.Logger) *ImageProxy { // from image.tmdb.org via their HTTP proxy without extra config. On // Windows we also honor the current user's system proxy settings. transport := NewExternalTransport() - if proxyConfiguredForImageFetch() { - // 走本地代理（如 127.0.0.1:7890）时，拨号目标是代理本身， - // 连接层 SSRF 校验会误杀本地回环代理；此时沿用 URL 级校验。 - log.Info("image proxy: outbound proxy detected, connection-level SSRF guard disabled") - } else { - // 仅 URL 解析层的 isPrivateHost 可被十进制/十六进制 IP、解析到 - // 私网的域名与 DNS rebinding 绕过；在拨号层对最终连接 IP 做二次 - // 校验（含重定向后的每条连接）堵住该旁路。 - // - // 用户明确配置的远程挂载源（如内网 Emby）与显式配置的服务（如 - // MetaTube）豁免该私网限制。豁免必须在 DialContext 这一层判断： - // Control 回调收到的 address 已经是**解析后的 IP**，而放行名单里 - // 存的是主机名（mebox-metatube-1、NAS 名），在 Control 里比较永远 - // 不相等——那段分支对主机名条目是死代码，现场表现就是「名单里有 - // 这个主机，回源依然被拒」。 - guardDialer := &net.Dialer{ - Timeout: 15 * time.Second, - Control: func(_, address string, _ syscall.RawConn) error { - host, _, err := net.SplitHostPort(address) - if err != nil { - return err - } - ip := net.ParseIP(host) - if ip == nil { - return errors.New("image proxy: refusing non-IP dial target") - } - if isPrivateIP(ip) { - return errors.New("image proxy: requests to private/internal hosts are not allowed") - } - return nil - }, - } - // 放行目标的专用拨号器：不再做私网校验，但仍然只连到请求里的那个 - // 主机（豁免的是私网限制，不是「任意目标」）。 - allowlistedDialer := &net.Dialer{Timeout: 15 * time.Second} - transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { - host, _, err := net.SplitHostPort(address) - if err != nil { - return nil, err - } - if proxy.isAllowedRemoteHost(host) { - return allowlistedDialer.DialContext(ctx, network, address) - } - return guardDialer.DialContext(ctx, network, address) - } - } - + proxy.guardTransport(transport) + directTransport := NewInternalTransport() + proxy.guardTransport(directTransport) proxy.client = &http.Client{Timeout: 30 * time.Second, Transport: transport} - proxy.directClient = &http.Client{Timeout: 30 * time.Second, Transport: NewInternalTransport()} + proxy.directClient = &http.Client{Timeout: 30 * time.Second, Transport: directTransport} return proxy } -// proxyConfiguredForImageFetch 探测环境变量或系统代理是否会影响图片抓取。 -func proxyConfiguredForImageFetch() bool { - req, err := http.NewRequest(http.MethodGet, "https://image.tmdb.org/", nil) - if err != nil { - return false - } - proxy, err := ProxyFromEnvironmentOrSystem(req) - return err == nil && proxy != nil -} - // SetLibraryRootsProvider injects a callback that returns the current set of // media library root directories. Sidecar posters live under these roots // (which are arbitrary, user-defined, and not necessarily under the diff --git a/internal/service/image_proxy_dial_allowlist_test.go b/internal/service/image_proxy_dial_allowlist_test.go index 10310ea..3d3c287 100644 --- a/internal/service/image_proxy_dial_allowlist_test.go +++ b/internal/service/image_proxy_dial_allowlist_test.go @@ -76,9 +76,37 @@ func TestImageProxyStillBlocksUnlistedHostnameResolvingToPrivateIP(t *testing.T) proxy.SetAllowedRemoteHostsProvider(func() []string { return nil }) _, port := hostAndPortOf(t, upstream.URL) - resp, err := proxy.client.Get("http://localhost:" + port + "/poster.jpg") - if err == nil { - resp.Body.Close() - t.Fatal("a non-allowlisted hostname resolving to a private IP must be refused") + for _, client := range []*http.Client{proxy.client, proxy.directClient} { + resp, err := client.Get("http://localhost:" + port + "/poster.jpg") + if err == nil { + resp.Body.Close() + t.Fatal("a non-allowlisted hostname resolving to a private IP must be refused") + } + } +} + +func TestImageProxyAllowsSelectedLocalProxy(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "image/jpeg") + _, _ = w.Write(testJPEG) + })) + defer upstream.Close() + proxyURL, err := url.Parse(upstream.URL) + if err != nil { + t.Fatal(err) + } + p := NewImageProxy(&config.Config{}, zap.NewNop()) + transport := NewExternalTransport() + transport.Proxy = http.ProxyURL(proxyURL) + p.guardTransport(transport) + t.Cleanup(transport.CloseIdleConnections) + client := &http.Client{Transport: transport} + resp, err := client.Get("http://images.example.com/poster.jpg") + if err != nil { + t.Fatalf("configured local proxy must remain usable: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status=%d, want 200", resp.StatusCode) } } diff --git a/internal/service/image_proxy_remote_fetch.go b/internal/service/image_proxy_remote_fetch.go index 3ae5f89..e59227e 100644 --- a/internal/service/image_proxy_remote_fetch.go +++ b/internal/service/image_proxy_remote_fetch.go @@ -32,7 +32,9 @@ func (p *ImageProxy) remoteImageFetchClients(host string) []remoteImageFetchClie } direct := p.directClient if direct == nil { - direct = &http.Client{Timeout: timeout, Transport: NewInternalTransport()} + transport := NewInternalTransport() + p.guardTransport(transport) + direct = &http.Client{Timeout: timeout, Transport: transport} } directCandidate := remoteImageFetchClient{name: "direct", client: direct} defaultCandidate := remoteImageFetchClient{name: "default", client: client} diff --git a/internal/service/image_proxy_transport.go b/internal/service/image_proxy_transport.go new file mode 100644 index 0000000..3999a4c --- /dev/null +++ b/internal/service/image_proxy_transport.go @@ -0,0 +1,67 @@ +package service + +import ( + "context" + "errors" + "net" + "net/http" + "net/url" + "strings" + "sync" + "syscall" + "time" +) + +// guardTransport checks resolved destination addresses even when a deployment +// has an outbound proxy. Only actual proxy endpoints and explicitly configured +// image services may use private addresses; NO_PROXY and direct fallback +// requests still need the connection-level check against DNS rebinding. +func (p *ImageProxy) guardTransport(transport *http.Transport) { + var proxyEndpoints sync.Map + if selectProxy := transport.Proxy; selectProxy != nil { + transport.Proxy = func(req *http.Request) (*url.URL, error) { + proxyURL, err := selectProxy(req) + if err == nil && proxyURL != nil { + port := proxyURL.Port() + if port == "" { + switch strings.ToLower(proxyURL.Scheme) { + case "https": + port = "443" + case "socks5", "socks5h": + port = "1080" + default: + port = "80" + } + } + proxyEndpoints.Store(net.JoinHostPort(strings.ToLower(proxyURL.Hostname()), port), true) + } + return proxyURL, err + } + } + guardDialer := &net.Dialer{ + Timeout: 15 * time.Second, + Control: func(_, address string, _ syscall.RawConn) error { + host, _, err := net.SplitHostPort(address) + if err != nil { + return err + } + ip := net.ParseIP(host) + if ip == nil || isPrivateIP(ip) { + return errors.New("image proxy: requests to private/internal hosts are not allowed") + } + return nil + }, + } + trustedDialer := &net.Dialer{Timeout: 15 * time.Second} + transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + host, _, err := net.SplitHostPort(address) + if err != nil { + return nil, err + } + _, isProxyEndpoint := proxyEndpoints.Load(strings.ToLower(address)) + if isProxyEndpoint || p.isAllowedRemoteHost(host) { + return trustedDialer.DialContext(ctx, network, address) + } + return guardDialer.DialContext(ctx, network, address) + } +} diff --git a/internal/service/transcoder_availability_test.go b/internal/service/transcoder_availability_test.go index fe89188..d554f35 100644 --- a/internal/service/transcoder_availability_test.go +++ b/internal/service/transcoder_availability_test.go @@ -13,6 +13,8 @@ import ( // be false both when transcoding is disabled by configuration and when no usable // ffmpeg can be resolved. func TestTranscoderAvailableGatesOnConfigAndBinary(t *testing.T) { + t.Setenv("PATH", t.TempDir()) + t.Chdir(t.TempDir()) cases := []struct { name string enabled bool diff --git a/web/src/pages/reader/ReaderReplacePage.tsx b/web/src/pages/reader/ReaderReplacePage.tsx index 8305323..59fda3f 100644 --- a/web/src/pages/reader/ReaderReplacePage.tsx +++ b/web/src/pages/reader/ReaderReplacePage.tsx @@ -35,7 +35,6 @@ export default function ReaderReplacePage() { .then(setRules) .catch(() => toast.error('加载替换规则失败')) } - // eslint-disable-next-line react-hooks/exhaustive-deps useEffect(load, []) const save = async () => { diff --git a/web/src/pages/reader/ReaderSourcesPage.tsx b/web/src/pages/reader/ReaderSourcesPage.tsx index e1c7522..be6e312 100644 --- a/web/src/pages/reader/ReaderSourcesPage.tsx +++ b/web/src/pages/reader/ReaderSourcesPage.tsx @@ -49,7 +49,6 @@ export default function ReaderSourcesPage() { .then(setSources) .catch(() => toast.error('加载书源失败')) } - // eslint-disable-next-line react-hooks/exhaustive-deps useEffect(load, []) // 读取书源文件（选择或拖入），走与粘贴相同的导入链路 diff --git a/web/src/pages/reader/ReaderViewPage.tsx b/web/src/pages/reader/ReaderViewPage.tsx index b67e86c..f591d14 100644 --- a/web/src/pages/reader/ReaderViewPage.tsx +++ b/web/src/pages/reader/ReaderViewPage.tsx @@ -1,4 +1,4 @@ -import { memo, useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState, type CSSProperties } from 'react' +import { memo, useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState, type CSSProperties, type ReactNode } from 'react' import { useParams } from 'react-router-dom' import toast from 'react-hot-toast' import { @@ -266,6 +266,19 @@ function settleTextDrag(dragEl: HTMLDivElement, turn: 1 | -1, dx: number, stride }) } +function ReaderMenuButton({ icon, label, onClick }: { icon: ReactNode; label: string; onClick: () => void }) { + return ( + + ) +} + export default function ReaderViewPage() { const { bookId = '' } = useParams() const settings = useReaderSettingsStore() @@ -316,6 +329,7 @@ export default function ReaderViewPage() { const [page, setPage] = useState(0) const [pageCount, setPageCount] = useState(1) const [vw, setVw] = useState(0) + const [scrollProgress, setScrollProgress] = useState(0) // 文本翻页：位移与页脚先走 DOM，React 的 page state 等到过渡结束后再同步。 // 若在同一次点击/触摸里同步 setPage，整页重渲染会堵住下一次绘制， // 命令式 transform 也要等卡完才开始——点按和滑动都会顿一下。 @@ -338,8 +352,13 @@ export default function ReaderViewPage() { const paragraphWarmRef = useRef(null) /** 当前窗口中心页；-1 表示窗口失效（映射刚重算过），下次应用时重建。 */ const warmCenterRef = useRef(-1) - vwRef.current = vw - pageCountRef.current = pageCount + const currentChapter = chapterIndex !== null ? chapters[chapterIndex] : null + const chapterTitle = currentChapter?.title ?? '' + useLayoutEffect(() => { + vwRef.current = vw + pageCountRef.current = pageCount + chapterTitleRef.current = chapterTitle + }, [vw, pageCount, chapterTitle]) // 音频/漫画媒体状态 const [media, setMedia] = useState(null) const [restorePos, setRestorePos] = useState(0) // 音频秒数 / 漫画图片序号 @@ -448,6 +467,7 @@ export default function ReaderViewPage() { // ── 加载书籍与章节 ── useEffect(() => { let cancelled = false + const cache = contentCache.current ;(async () => { try { setLoadingStage('book') @@ -510,7 +530,7 @@ export default function ReaderViewPage() { })() return () => { cancelled = true - contentCache.current.clear() + cache.clear() } }, [bookId, reloadKey]) @@ -910,6 +930,16 @@ export default function ReaderViewPage() { if (menuOpen) flushTextPageReactSync() }, [menuOpen, flushTextPageReactSync]) + // Only refresh the scroll slider while its menu is visible, keeping scroll + // events on the reading path free of whole-page React renders. + useLayoutEffect(() => { + if (!menuOpen || settings.pageMode !== 'scroll') return + const el = scrollRef.current + if (el) { + setScrollProgress(Math.round((el.scrollTop / Math.max(1, el.scrollHeight - el.clientHeight)) * 1000)) + } + }, [menuOpen, settings.pageMode, content]) + // 滚动模式恢复进度。 // // 同时清掉 dataset.last：它是「上次保存的滚动百分比」，换章后必须重算。 @@ -1355,9 +1385,6 @@ export default function ReaderViewPage() { return () => el.removeEventListener('wheel', onWheel) }, [settings.pageMode, menuOpen, contentType, goPrev, goNext]) - const currentChapter = chapterIndex !== null ? chapters[chapterIndex] : null - chapterTitleRef.current = currentChapter?.title ?? '' - // 顶栏高度量一次：菜单打开时正文下移，开头几行不被顶栏压住。 // 用 transform 而不是 padding，避免改变视口高度触发重新分页。 const menuShiftStyle = useMemo( @@ -1397,11 +1424,7 @@ export default function ReaderViewPage() { value: settings.pageMode === 'page' ? page + 1 - : Math.round( - ((scrollRef.current?.scrollTop ?? 0) / - Math.max(1, (scrollRef.current?.scrollHeight ?? 1) - (scrollRef.current?.clientHeight ?? 1))) * - 1000, - ), + : scrollProgress, onChange: (v: number) => { if (settings.pageMode === 'page') { commitTextPage(v - 1) @@ -1569,6 +1592,7 @@ export default function ReaderViewPage() { const max = el.scrollHeight - el.clientHeight if (max > 0) { const pct = el.scrollTop / max + if (menuOpen) setScrollProgress(Math.round(pct * 1000)) if (shouldSaveScrollProgress(pct, el.dataset.last)) { el.dataset.last = formatScrollPct(pct) savePos(Math.round(el.scrollTop)) @@ -1753,62 +1777,35 @@ export default function ReaderViewPage() { 用 flex 等分而不是 grid-cols-N：动作数量随内容类型和书源能力变化， grid 列数得写死成字面量类名，多一个按钮就要再加一档。 */}
- {([ - { icon: , label: '目录', action: () => openPanel('toc') }, - { icon: , label: '界面', action: () => openPanel('style') }, - { - icon: settings.night ? : , - label: settings.night ? '日间' : '夜间', - action: () => settings.toggleNight(), - }, - { - icon: settings.pageMode === 'page' ? : , - label: settings.pageMode === 'page' ? '滚动' : '翻页', - action: () => settings.setPageMode(settings.pageMode === 'page' ? 'scroll' : 'page'), - }, - // 双页铺开只对漫画的翻页模式有意义；窗口太窄时并排两页没法看， - // 干脆不显示入口（不是灰掉，避免用户以为坏了）。 - ...(contentType === 'image' && settings.pageMode === 'page' && windowWidth >= COMIC_SPREAD_MIN_WIDTH - ? [ - { - icon: , - label: settings.comicDoublePage ? '单页' : '双页', - action: () => settings.setComicDoublePage(!settings.comicDoublePage), - }, - ] - : []), - // 全屏：只在浏览器支持元素全屏时出现（iOS Safari 不支持，见上方注释）。 - ...(fullscreenSupported - ? [ - { - icon: fullscreen ? : , - label: fullscreen ? '还原' : '全屏', - action: toggleFullscreen, - }, - ] - : []), - ...(canSwitchSource - ? [ - { - icon: , - label: '换源', - action: () => { - void openSourcePicker() - }, - }, - ] - : []), - ]).map((item) => ( - - ))} + } label="目录" onClick={() => openPanel('toc')} /> + } label="界面" onClick={() => openPanel('style')} /> + : } + label={settings.night ? '日间' : '夜间'} + onClick={() => settings.toggleNight()} + /> + : } + label={settings.pageMode === 'page' ? '滚动' : '翻页'} + onClick={() => settings.setPageMode(settings.pageMode === 'page' ? 'scroll' : 'page')} + /> + {contentType === 'image' && settings.pageMode === 'page' && windowWidth >= COMIC_SPREAD_MIN_WIDTH && ( + } + label={settings.comicDoublePage ? '单页' : '双页'} + onClick={() => settings.setComicDoublePage(!settings.comicDoublePage)} + /> + )} + {fullscreenSupported && ( + : } + label={fullscreen ? '还原' : '全屏'} + onClick={toggleFullscreen} + /> + )} + {canSwitchSource && ( + } label="换源" onClick={() => { void openSourcePicker() }} /> + )}
{/* 界面设置面板（主题 / 字号 / 行距 / 段距 / 漫画图片尺寸） */} From 27dcfc2fe91dc4e192be9f74821a5a1988540548 Mon Sep 17 00:00:00 2001 From: jason jia Date: Sat, 17 Oct 2026 01:30:04 +0000 Subject: [PATCH 2/2] fix(playback): reuse completed HLS output and preserve resume positions --- internal/service/transcoder.go | 73 ++++-- internal/service/transcoder_ffmpeg_probe.go | 90 ++++--- internal/service/transcoder_jobs.go | 79 ++++-- internal/service/transcoder_lifecycle_test.go | 228 ++++++++++++++++++ web/src/api/playback.ts | 4 +- web/src/pages/PlayerPage.tsx | 108 ++++----- web/src/pages/hlsPlayback.test.ts | 58 +++++ web/src/pages/hlsPlayback.ts | 43 ++++ 8 files changed, 557 insertions(+), 126 deletions(-) create mode 100644 internal/service/transcoder_lifecycle_test.go create mode 100644 web/src/pages/hlsPlayback.test.ts create mode 100644 web/src/pages/hlsPlayback.ts diff --git a/internal/service/transcoder.go b/internal/service/transcoder.go index 4c0e7d6..357981a 100644 --- a/internal/service/transcoder.go +++ b/internal/service/transcoder.go @@ -59,6 +59,9 @@ type TranscoderService struct { availMu sync.Mutex availCheckedAt time.Time availValue bool + availKey string + ffmpegMu sync.Mutex + ffmpegCache ffmpegValidationCache } // hlsJob holds the live state of one ffmpeg run. @@ -83,6 +86,11 @@ type hlsJob struct { quality string // done is closed when the ffmpeg goroutine fully exits (after process death). done chan struct{} + // Completed outputs stay reusable until idle eviction without occupying a + // running-process slot. The binary and config are fixed for this generation. + completed bool + ffmpegPath string + config *config.Config } var ( @@ -117,24 +125,37 @@ func (t *TranscoderService) Available() bool { if t == nil || t.cfg == nil { return false } - if !t.cfg.Transcoder.Enabled { + cfg := t.transcodeConfig() + if !cfg.Transcoder.Enabled { return false } + encoder := ffmpegArgSettingsFromConfig(cfg, nil).encoder + key := cfg.App.FFmpegPath + "\x00" + encoder t.availMu.Lock() defer t.availMu.Unlock() - if !t.availCheckedAt.IsZero() && time.Since(t.availCheckedAt) < transcodeAvailabilityTTL { - return t.availValue + if !t.availValue && t.availKey == key && !t.availCheckedAt.IsZero() && time.Since(t.availCheckedAt) < transcodeAvailabilityTTL { + return false } - _, err := t.resolveFFmpegPath() + _, err := t.resolveFFmpegPathFor(context.Background(), cfg.App.FFmpegPath, encoder) available := err == nil if !available && t.log != nil { t.log.Debug("local HLS renditions unavailable", zap.Error(err)) } t.availValue = available t.availCheckedAt = time.Now() + t.availKey = key return available } +func (t *TranscoderService) transcodeConfig() *config.Config { + config.RuntimeMu.RLock() + defer config.RuntimeMu.RUnlock() + return &config.Config{ + App: config.AppConfig{FFmpegPath: t.cfg.App.FFmpegPath, VAAPIDevice: t.cfg.App.VAAPIDevice}, + Transcoder: t.cfg.Transcoder, + } +} + // HLSDir is the per-media directory that holds index.m3u8 + segment files. func (t *TranscoderService) HLSDir(mediaID string) string { return filepath.Join(t.cfg.Cache.CacheDir, "hls", mediaID) @@ -174,7 +195,8 @@ func (t *TranscoderService) EnsureJobFromSubtitle(ctx context.Context, mediaID s // EnsureJobFromSubtitleQuality is EnsureJobFromSubtitle with an explicit local // HLS quality profile. Empty qualityID keeps the global transcoder config. func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, mediaID string, startSec float64, seekGen int64, subtitleStream int, qualityID string) (string, error) { - if !t.cfg.Transcoder.Enabled { + cfg := t.transcodeConfig() + if !cfg.Transcoder.Enabled { return "", ErrTranscodeDisabled } if startSec < 0 { @@ -203,12 +225,17 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me t.mu.Lock() if existing, ok := t.jobs[mediaID]; ok { - if sameHLSConfiguration(existing, startSec, subtitleStream, qualityID) { + reusable := true + if existing.completed { + info, err := os.Stat(t.PlaylistPath(mediaID)) + reusable = err == nil && info.Size() > 0 + } + if reusable && sameHLSConfiguration(existing, startSec, subtitleStream, qualityID) { t.touchJobLocked(mediaID) t.mu.Unlock() return t.PlaylistPath(mediaID), nil } - if !shouldReplaceHLSJobConfigurationQuality(existing, startSec, seekGen, subtitleStream, qualityID) { + if reusable && !shouldReplaceHLSJobConfigurationQuality(existing, startSec, seekGen, subtitleStream, qualityID) { t.touchJobLocked(mediaID) t.mu.Unlock() return t.PlaylistPath(mediaID), nil @@ -230,7 +257,13 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me } input.Quality = quality t.maybeFillDuration(ctx, m, input) - if _, err := t.resolveFFmpegPath(); err != nil { + settings := ffmpegArgSettingsFromConfig(cfg, quality) + jobEncoder := settings.encoder + if settings.noScale || subtitleStream >= 0 { + jobEncoder = "" + } + bin, err := t.resolveFFmpegPathFor(ctx, cfg.App.FFmpegPath, jobEncoder) + if err != nil { return "", err } @@ -242,29 +275,15 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me } t.mu.Lock() - if existing, ok := t.jobs[mediaID]; ok { - if sameHLSConfiguration(existing, startSec, subtitleStream, qualityID) { - t.touchJobLocked(mediaID) - t.mu.Unlock() - return t.PlaylistPath(mediaID), nil - } - if !shouldReplaceHLSJobConfigurationQuality(existing, startSec, seekGen, subtitleStream, qualityID) { - t.touchJobLocked(mediaID) - t.mu.Unlock() - return t.PlaylistPath(mediaID), nil - } - prev := t.detachJobLocked(mediaID) - t.mu.Unlock() - waitJobExit(prev, 12*time.Second) - t.mu.Lock() - } - if max := t.maxConcurrent(); max > 0 && len(t.jobs) >= max { + if max := t.maxConcurrent(); max > 0 && t.runningJobCountLocked() >= max { t.mu.Unlock() return "", ErrTranscodeBusy } jobCtx, cancel := context.WithCancel(context.Background()) - jobEncoder := t.effectiveEncoder() + if jobEncoder == "" { + jobEncoder = "software" + } if subtitleStream >= 0 { jobEncoder = "software (subtitle burn-in)" } @@ -280,6 +299,8 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me subtitleStream: subtitleStream, quality: qualityID, done: make(chan struct{}), + ffmpegPath: bin, + config: cfg, } t.jobs[mediaID] = job t.mu.Unlock() diff --git a/internal/service/transcoder_ffmpeg_probe.go b/internal/service/transcoder_ffmpeg_probe.go index dee811b..9cc6c1c 100644 --- a/internal/service/transcoder_ffmpeg_probe.go +++ b/internal/service/transcoder_ffmpeg_probe.go @@ -11,29 +11,17 @@ import ( "time" "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" ) func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input transcodeInput) { - bin, err := t.resolveFFmpegPath() - if err != nil { - t.log.Warn("ffmpeg unavailable", zap.String("media_id", job.mediaID), zap.Error(err)) - t.mu.Lock() - delete(t.jobs, job.mediaID) - t.mu.Unlock() - t.hub.Publish("transcode", map[string]any{ - "media_id": job.mediaID, - "status": "error", - "error": err.Error(), - }) - return - } - playlist := filepath.Join(job.outputDir, "index.m3u8") segments := filepath.Join(job.outputDir, "seg_%05d.ts") - args := buildFFmpegArgsForInput(t.cfg, input, playlist, segments) + args := buildFFmpegArgsForInput(job.config, input, playlist, segments) - cmd := exec.CommandContext(ctx, bin, args...) // #nosec G204 -- bin is resolved by resolveFFmpegPath and args are passed without a shell. + cmd := exec.CommandContext(ctx, job.ffmpegPath, args...) // #nosec G204 -- the binary was validated before creating this job. setFFmpegSysProcAttr(cmd) cmd.Stderr = os.Stderr @@ -50,7 +38,8 @@ func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input tr "start_sec": input.StartSec, }) - if err := cmd.Run(); err != nil && !errors.Is(ctx.Err(), context.Canceled) { + err := cmd.Run() + if err != nil && !errors.Is(ctx.Err(), context.Canceled) { t.log.Warn("ffmpeg exited", zap.String("media_id", job.mediaID), zap.Float64("start_sec", input.StartSec), @@ -59,11 +48,18 @@ func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input tr } t.mu.Lock() - // Only drop the map entry if we are still the registered generation. + // Only update the map entry if we are still the registered generation. if cur, ok := t.jobs[job.mediaID]; ok && cur == job { - delete(t.jobs, job.mediaID) + if err == nil && ctx.Err() == nil { + job.completed = true + } else { + delete(t.jobs, job.mediaID) + } } t.mu.Unlock() + if err != nil || ctx.Err() != nil { + job.cancel() + } t.hub.Publish("transcode", map[string]any{ "media_id": job.mediaID, @@ -73,19 +69,53 @@ func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input tr } func (t *TranscoderService) resolveFFmpegPath() (string, error) { - t.mu.Lock() - configuredPath := strings.TrimSpace(t.cfg.App.FFmpegPath) - t.mu.Unlock() + cfg := t.transcodeConfig() + return t.resolveFFmpegPathFor(context.Background(), cfg.App.FFmpegPath, ffmpegArgSettingsFromConfig(cfg, nil).encoder) +} + +type ffmpegValidationCache struct { + configuredPath string + path string + encoder string + size int64 + modTime time.Time + checkedAt time.Time +} + +func (t *TranscoderService) resolveFFmpegPathFor(ctx context.Context, configuredPath, encoder string) (string, error) { + t.ffmpegMu.Lock() + defer t.ffmpegMu.Unlock() + if err := ctx.Err(); err != nil { + return "", err + } + configuredPath = strings.TrimSpace(configuredPath) + cached := t.ffmpegCache + if cached.path != "" && cached.encoder == encoder && + (configuredPath == cached.configuredPath || configuredPath == cached.path) && + time.Since(cached.checkedAt) < transcodeAvailabilityTTL { + if info, err := os.Stat(cached.path); err == nil && info.Size() == cached.size && info.ModTime().Equal(cached.modTime) { + return cached.path, nil + } + } var lastErr error for _, bin := range executableCandidates(configuredPath, "ffmpeg") { - if err := validateFFmpegForTranscode(context.Background(), bin, t.effectiveEncoder()); err != nil { + if err := validateFFmpegForTranscode(ctx, bin, encoder); err != nil { + if ctx.Err() != nil { + return "", ctx.Err() + } lastErr = err continue } - t.mu.Lock() - t.cfg.App.FFmpegPath = bin - t.mu.Unlock() + if info, err := os.Stat(bin); err == nil { + t.ffmpegCache = ffmpegValidationCache{configuredPath: configuredPath, path: bin, encoder: encoder, + size: info.Size(), modTime: info.ModTime(), checkedAt: time.Now()} + } + config.RuntimeMu.Lock() + if strings.TrimSpace(t.cfg.App.FFmpegPath) == configuredPath { + t.cfg.App.FFmpegPath = bin + } + config.RuntimeMu.Unlock() return bin, nil } if lastErr != nil { @@ -95,10 +125,11 @@ func (t *TranscoderService) resolveFFmpegPath() (string, error) { } func (t *TranscoderService) effectiveEncoder() string { - if !t.cfg.Transcoder.HardwareAccel { + cfg := t.transcodeConfig() + if !cfg.Transcoder.HardwareAccel { return "" } - return normalizedHardwareEncoder(t.cfg.Transcoder.Encoder) + return normalizedHardwareEncoder(cfg.Transcoder.Encoder) } func normalizedHardwareEncoder(encoder string) string { @@ -160,6 +191,7 @@ func hasFFmpegListEntry(output, name string) bool { // HumanFFmpegProfile is exposed for the admin UI / settings view. func (t *TranscoderService) HumanFFmpegProfile() string { + cfg := t.transcodeConfig() return fmt.Sprintf("ffmpeg=%s, encoder=%s, output=%s", - t.cfg.App.FFmpegPath, t.cfg.Transcoder.Encoder, filepath.Join(t.cfg.Cache.CacheDir, "hls")) + cfg.App.FFmpegPath, cfg.Transcoder.Encoder, filepath.Join(t.cfg.Cache.CacheDir, "hls")) } diff --git a/internal/service/transcoder_jobs.go b/internal/service/transcoder_jobs.go index daedb3e..496ffdb 100644 --- a/internal/service/transcoder_jobs.go +++ b/internal/service/transcoder_jobs.go @@ -6,41 +6,58 @@ import ( "time" "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" ) // WaitReady blocks (with a deadline) until the playlist file shows up on // disk for the *current* job generation. Stale playlists left behind by a // failed RemoveAll / still-exiting ffmpeg must not unblock a mid-file restart. func (t *TranscoderService) WaitReady(ctx context.Context, mediaID string, timeout time.Duration) bool { - deadline := time.Now().Add(timeout) + timer := time.NewTimer(timeout) + defer timer.Stop() + ticker := time.NewTicker(250 * time.Millisecond) + defer ticker.Stop() for { + if ctx.Err() != nil { + return false + } t.mu.Lock() job, ok := t.jobs[mediaID] var started time.Time + var completed bool if ok { started = job.startedAt + completed = job.completed } t.mu.Unlock() + if !ok { + return false // A failed/exited job cannot become ready later. + } if ok { if info, err := os.Stat(t.PlaylistPath(mediaID)); err == nil { // Allow a small clock skew; reject anything older than this job. if !info.ModTime().Before(started.Add(-2 * time.Second)) { t.mu.Lock() - if j, exists := t.jobs[mediaID]; exists { + if j, exists := t.jobs[mediaID]; exists && j == job { j.playlistOK = true + t.mu.Unlock() + return true } t.mu.Unlock() - return true } } } - if time.Now().After(deadline) || ctx.Err() != nil { - return false + if completed { + return false // A completed job cannot regenerate a missing playlist. } select { case <-ctx.Done(): return false - case <-time.After(250 * time.Millisecond): + case <-timer.C: + return false + case <-ticker.C: + case <-job.done: } } } @@ -101,6 +118,9 @@ func (t *TranscoderService) Active() []ActiveJob { defer t.mu.Unlock() out := make([]ActiveJob, 0, len(t.jobs)) for _, j := range t.jobs { + if j.completed { + continue + } out = append(out, ActiveJob{ MediaID: j.mediaID, Encoder: j.encoder, @@ -111,7 +131,19 @@ func (t *TranscoderService) Active() []ActiveJob { return out } +func (t *TranscoderService) runningJobCountLocked() int { + n := 0 + for _, job := range t.jobs { + if !job.completed { + n++ + } + } + return n +} + func (t *TranscoderService) maxConcurrent() int { + config.RuntimeMu.RLock() + defer config.RuntimeMu.RUnlock() if t.cfg.Transcoder.MaxConcurrent <= 0 { return 1 } @@ -119,6 +151,8 @@ func (t *TranscoderService) maxConcurrent() int { } func (t *TranscoderService) idleTimeout() time.Duration { + config.RuntimeMu.RLock() + defer config.RuntimeMu.RUnlock() if t.cfg.Transcoder.IdleTimeoutSeconds <= 0 { return 120 * time.Second } @@ -137,21 +171,38 @@ func (t *TranscoderService) monitorIdle(ctx context.Context, job *hlsJob) { case <-ticker.C: t.mu.Lock() current, ok := t.jobs[job.mediaID] - if !ok { + if !ok || current != job { t.mu.Unlock() return } idleFor := time.Since(current.lastAccess) t.mu.Unlock() - if idleFor >= timeout { - t.log.Info("transcode idle timeout", - zap.String("media_id", job.mediaID), - zap.Duration("idle_for", idleFor), - zap.Duration("timeout", timeout), - ) - t.StopJob(job.mediaID) + if idleFor >= timeout && t.stopIdleJob(job, timeout) { return } } } } + +func (t *TranscoderService) stopIdleJob(job *hlsJob, timeout time.Duration) bool { + gate := t.mediaStartGate(job.mediaID) + gate.Lock() + defer gate.Unlock() + t.mu.Lock() + current := t.jobs[job.mediaID] + if current != job { + t.mu.Unlock() + return true + } + idleFor := time.Since(job.lastAccess) + if idleFor < timeout { + t.mu.Unlock() + return false + } + t.detachJobLocked(job.mediaID) + t.mu.Unlock() + waitJobExit(job, 12*time.Second) + t.log.Info("transcode idle timeout", zap.String("media_id", job.mediaID), + zap.Duration("idle_for", idleFor), zap.Duration("timeout", timeout)) + return true +} diff --git a/internal/service/transcoder_lifecycle_test.go b/internal/service/transcoder_lifecycle_test.go new file mode 100644 index 0000000..4adb7cb --- /dev/null +++ b/internal/service/transcoder_lifecycle_test.go @@ -0,0 +1,228 @@ +package service + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "testing" + "time" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/repository" +) + +func TestHLSCompletedTranscodeStaysReadyAndIsReused(t *testing.T) { + bin, err := exec.LookPath("ffmpeg") + if err != nil { + t.Skip("ffmpeg is required for the real HLS lifecycle test") + } + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + source := filepath.Join(t.TempDir(), "short.mp4") + cmd := exec.CommandContext(ctx, bin, "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "testsrc=size=64x64:rate=10", "-t", "1", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-threads", "1", source) + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("generate source: %v: %s", err, output) + } + cfg := &config.Config{} + cfg.App.FFmpegPath = bin + cfg.Cache.CacheDir = t.TempDir() + cfg.Transcoder.Enabled = true + cfg.Transcoder.MaxConcurrent = 1 + cfg.Transcoder.MaxHeight = 64 + cfg.Transcoder.SegmentSeconds = 1 + cfg.Transcoder.Threads = 1 + db := newServiceTestDB(t) + media := model.Media{Path: source, DurationSec: 1} + if err := db.Create(&media).Error; err != nil { + t.Fatal(err) + } + svc := NewTranscoderService(cfg, zap.NewNop(), repository.New(db), NewHub(zap.NewNop())) + t.Cleanup(svc.StopAll) + playlist, err := svc.EnsureJob(ctx, media.ID) + if err != nil { + t.Fatal(err) + } + svc.mu.Lock() + job := svc.jobs[media.ID] + svc.mu.Unlock() + if job == nil { + t.Fatal("transcode lost its job before the playlist could be read") + } + select { + case <-job.done: + case <-ctx.Done(): + t.Fatal("short transcode did not finish") + } + data, err := os.ReadFile(playlist) + if err != nil || !strings.Contains(string(data), "#EXT-X-ENDLIST") { + t.Fatalf("transcode did not produce a complete playlist: err=%v, data=%s", err, data) + } + if !svc.WaitReady(ctx, media.ID, 100*time.Millisecond) { + t.Fatal("completed HLS playlist is incorrectly reported as not ready") + } + if len(svc.Active()) != 0 { + t.Fatal("completed transcode still occupies an active job slot") + } + if _, err := svc.EnsureJob(ctx, media.ID); err != nil { + t.Fatal(err) + } + svc.mu.Lock() + reused := svc.jobs[media.ID] == job + svc.mu.Unlock() + if !reused { + t.Fatal("refreshing a completed playlist started a second transcode") + } + other := model.Media{Path: source + ".second", DurationSec: 1} + if err := os.WriteFile(other.Path, mustReadTestFile(t, source), 0o600); err != nil { + t.Fatal(err) + } + if err := db.Create(&other).Error; err != nil { + t.Fatal(err) + } + if _, err := svc.EnsureJob(ctx, other.ID); err != nil { + t.Fatalf("completed cache must not consume MaxConcurrent: %v", err) + } +} + +func mustReadTestFile(t *testing.T, path string) []byte { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return data +} + +func TestHLSWaitReadyReturnsImmediatelyForMissingJob(t *testing.T) { + svc := NewTranscoderService(&config.Config{}, zap.NewNop(), nil, NewHub(zap.NewNop())) + started := time.Now() + if svc.WaitReady(t.Context(), "failed", time.Second) { + t.Fatal("missing job must not be ready") + } + if elapsed := time.Since(started); elapsed >= 500*time.Millisecond { + t.Fatalf("missing/failed job waited for the readiness deadline: %v", elapsed) + } +} + +func countingFFmpeg(t *testing.T) string { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("test ffmpeg uses a POSIX shell") + } + bin := filepath.Join(t.TempDir(), "ffmpeg") + script := `#!/bin/sh +printf '%s\n' "$*" >> "$0.calls" +case " $* " in + *" -encoders "*) printf ' V..... libx264\n V..... h264_nvenc\n' ;; + *" -muxers "*) printf ' E hls\n' ;; +esac +` + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + return bin +} + +func TestFFmpegValidationIsCachedAndInvalidatedByEncoder(t *testing.T) { + bin := countingFFmpeg(t) + cfg := &config.Config{} + cfg.App.FFmpegPath = bin + svc := NewTranscoderService(cfg, zap.NewNop(), nil, NewHub(zap.NewNop())) + for range 3 { + if _, err := svc.resolveFFmpegPath(); err != nil { + t.Fatal(err) + } + } + calls := strings.Count(string(mustReadTestFile(t, bin+".calls")), "\n") + if calls != 2 { + t.Fatalf("repeated resolution spawned ffmpeg %d times, want one encoder/muxer probe pair", calls) + } + ApplyRuntimeSetting(cfg, "transcoder.hardware_accel", "true") + ApplyRuntimeSetting(cfg, "transcoder.encoder", "nvenc") + if _, err := svc.resolveFFmpegPath(); err != nil { + t.Fatal(err) + } + calls = strings.Count(string(mustReadTestFile(t, bin+".calls")), "\n") + if calls != 4 { + t.Fatalf("encoder change must revalidate capabilities, calls=%d", calls) + } + changed := time.Now().Add(time.Second) + if err := os.Chtimes(bin, changed, changed); err != nil { + t.Fatal(err) + } + if _, err := svc.resolveFFmpegPath(); err != nil { + t.Fatal(err) + } + calls = strings.Count(string(mustReadTestFile(t, bin+".calls")), "\n") + if calls != 6 { + t.Fatalf("binary replacement must invalidate validation, calls=%d", calls) + } +} + +func TestHLSIdleCleanupPreservesReplacementAndRecentActivity(t *testing.T) { + svc := NewTranscoderService(&config.Config{}, zap.NewNop(), nil, NewHub(zap.NewNop())) + done := make(chan struct{}) + close(done) + cancelled := false + old := &hlsJob{mediaID: "m", lastAccess: time.Now().Add(-time.Hour)} + current := &hlsJob{mediaID: "m", lastAccess: time.Now(), done: done, cancel: func() { cancelled = true }} + svc.jobs["m"] = current + if !svc.stopIdleJob(old, time.Minute) || cancelled || svc.jobs["m"] != current { + t.Fatal("an old watchdog must leave the replacement generation intact") + } + if svc.stopIdleJob(current, time.Minute) || cancelled { + t.Fatal("recent segment access must prevent idle cancellation") + } + current.lastAccess = time.Now().Add(-time.Hour) + if !svc.stopIdleJob(current, time.Minute) || !cancelled || svc.jobs["m"] != nil { + t.Fatal("an idle generation must release its resources") + } +} + +func TestHLSOriginalQualityValidatesSoftwareEncoder(t *testing.T) { + bin := countingFFmpeg(t) + softwareOnly := strings.ReplaceAll(string(mustReadTestFile(t, bin)), ` V..... h264_nvenc\n`, "") + if err := os.WriteFile(bin, []byte(softwareOnly), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", t.TempDir()) + t.Chdir(t.TempDir()) + cfg := &config.Config{} + cfg.App.FFmpegPath = bin + cfg.Transcoder.Enabled = true + cfg.Transcoder.HardwareAccel = true + cfg.Transcoder.Encoder = "nvenc" + cfg.Cache.CacheDir = t.TempDir() + db := newServiceTestDB(t) + source := filepath.Join(t.TempDir(), "movie.mkv") + if err := os.WriteFile(source, []byte("source"), 0o600); err != nil { + t.Fatal(err) + } + media := model.Media{Path: source, DurationSec: 60} + if err := db.Create(&media).Error; err != nil { + t.Fatal(err) + } + svc := NewTranscoderService(cfg, zap.NewNop(), repository.New(db), NewHub(zap.NewNop())) + t.Cleanup(svc.StopAll) + if _, err := svc.EnsureJobFromSubtitleQuality(t.Context(), media.ID, 0, 1, -1, "source"); err != nil { + t.Fatalf("source quality uses software and must not require NVENC: %v", err) + } + svc.mu.Lock() + job := svc.jobs[media.ID] + svc.mu.Unlock() + select { + case <-job.done: + case <-time.After(time.Second): + t.Fatal("fake ffmpeg did not finish") + } + calls := string(mustReadTestFile(t, bin+".calls")) + if strings.Count(calls, "\n") != 3 || !strings.Contains(calls, "-c:v libx264") { + t.Fatalf("startup must probe once and encode with libx264, calls=%s", calls) + } +} diff --git a/web/src/api/playback.ts b/web/src/api/playback.ts index add8ced..2138b8e 100644 --- a/web/src/api/playback.ts +++ b/web/src/api/playback.ts @@ -42,9 +42,9 @@ function publicOriginHeader() { } export const playbackAPI = { - getResume: (mediaId: string) => + getResume: (mediaId: string, signal?: AbortSignal) => api - .get<{ position_ms: number; duration_ms: number; completed: boolean }>(`/playback/${mediaId}/resume`) + .get<{ position_ms: number; duration_ms: number; completed: boolean }>(`/playback/${encodeURIComponent(mediaId)}/resume`, { signal }) .then((r) => r.data), // 片头/片尾片段：播放开始后再调用，服务端可能需要几秒去外部数据库取数， diff --git a/web/src/pages/PlayerPage.tsx b/web/src/pages/PlayerPage.tsx index c3650b2..17a1d16 100644 --- a/web/src/pages/PlayerPage.tsx +++ b/web/src/pages/PlayerPage.tsx @@ -40,6 +40,7 @@ import { pickPlayerMode, needsTranscodeForBrowser, isDirectStreamMedia, isStrmMe import { classifyDirectPlayError, directPlayFailureCopy, directStallTimeoutMs, DIRECT_SEEK_GRACE_MS } from './directPlayError' import type { DirectPlayFailureReason } from './directPlayError' import { canSeekDirectNow, DIRECT_SEEK_WAIT_MS } from './directPlaySeek' +import { applyPendingHlsSeek, attachNativeHls, bindPendingHlsSeek, hlsManifestTimeoutMs, playbackPositionSec } from './hlsPlayback' import { apiErrorMessage } from './StrmManagePage' import { PlayerTopBar } from './PlayerTopBar' import { PlayerVideoStage } from './PlayerVideoStage' @@ -992,14 +993,15 @@ export function PlayerPage() { }, []) const switchToLocalHLS = useCallback( - (position = 0) => { + (position?: number) => { + pendingSeekRef.current = null setHlsSource('local') setCloudWaiting(false) setSelectedQuality((prev) => { if (playbackInfo?.local_qualities?.some((quality) => quality.id === prev)) return prev return defaultLocalQualityId(playbackInfo) }) - if (position > 2) setHlsStartSec(position) + if (position !== undefined) setHlsStartSec(Math.max(0, position)) setPlaybackMode('hls') }, [playbackInfo, setPlaybackMode], @@ -1076,10 +1078,10 @@ export function PlayerPage() { : undefined if (preferred?.source === 'cloud') { setHlsSource('cloud') + // 云 HLS 是整段时间轴，等待云转码期间也要保留目标位置。 + pendingSeekRef.current = position > 2 ? position : null if (preferred.available) { setCloudWaiting(false) - // 云 HLS 是整段时间轴：切过去后按当前位置跳一下，别让用户从头看。 - pendingSeekRef.current = position > 2 ? position : null } else { setCloudWaiting(true) setCloudWaitMessage(preferred.note || `正在等待 115 转码 ${preferred.label}…`) @@ -1144,8 +1146,9 @@ export function PlayerPage() { if (!playbackInfo || !initialHlsRequestRef.current) return if (directOnly || !mediaId) return initialHlsRequestRef.current = false - enterHlsPlayback(0, playbackInfo) - }, [directOnly, enterHlsPlayback, mediaId, playbackInfo]) + const position = Math.max(resumePosition, playbackPositionSec(mode, hlsSource, hlsStartSec, ref.current?.currentTime || 0)) + enterHlsPlayback(position, playbackInfo) + }, [directOnly, enterHlsPlayback, hlsSource, hlsStartSec, mediaId, mode, playbackInfo, resumePosition]) // 直连播放只发现外挂字幕；只有 HLS 模式需要探测可烧录的内嵌字幕。 useEffect(() => { @@ -1294,6 +1297,8 @@ export function PlayerPage() { const currentMedia = mediaRef.current if (!currentMedia) return let cancelled = false + let cleanupHlsResume: (() => void) | undefined + let onDirectMetadata: (() => void) | undefined teardownHls() const video = ref.current @@ -1310,13 +1315,14 @@ export function PlayerPage() { void import('hls.js').then(({ default: HlsCtor }) => { if (cancelled || !ref.current) return if (HlsCtor.isSupported()) { + cleanupHlsResume = bindPendingHlsSeek(video, pendingSeekRef) const hls = new HlsCtor({ enableWorker: true, lowLatencyMode: false, // The server waits up to 45s for the first segment. Slow two-core // hosts and remote STRM sources regularly need more than hls.js's // 10s default, which otherwise aborts a healthy transcode. - manifestLoadingTimeOut: 60_000, + manifestLoadingTimeOut: hlsManifestTimeoutMs(hlsSource, hlsStartSec), // 115 云 HLS 的 master.m3u8 每次都要现调 115 接口（实测 0.5–1.3s）， // 还可能撞上 409「正在转码」。只重试 1 次的话，一次抖动就判死刑， // 直接掉到本地转码。 @@ -1330,15 +1336,7 @@ export function PlayerPage() { hls.loadSource(url) hls.attachMedia(video) hls.on(HlsCtor.Events.MANIFEST_PARSED, () => { - if (pendingSeekRef.current !== null) { - const target = pendingSeekRef.current - pendingSeekRef.current = null - try { - video.currentTime = target - } catch { - // ignore - } - } + applyPendingHlsSeek(video, pendingSeekRef) void video.play().catch(() => undefined) // .strm 入库时常缺 duration；转码启动时会补探测，这里刷新一次给进度条总时长。 if (durationSec > 0) return @@ -1388,8 +1386,7 @@ export function PlayerPage() { } catch { // ignore } - video.src = url - void video.play().catch(() => undefined) + cleanupHlsResume = attachNativeHls(video, url, pendingSeekRef) } else { setHlsUnavailable(true) showPlayFailure('当前浏览器不支持 HLS，正在尝试直接播放。') @@ -1421,23 +1418,20 @@ export function PlayerPage() { ) pendingDirectStartRef.current = 0 if (resumeAt > 0) { - video.addEventListener( - 'loadedmetadata', - () => { - // STRM/115 直链要等 seekable 覆盖目标位置再写 currentTime（否则会被 - // 钳回 0），并顺带打开误报宽限期，避免这次跳转被当成播放失败又弹回 HLS。 - if (isStrmMedia(currentMedia)) { - seekStrmDirectTo(resumeAt) - return - } - try { - video.currentTime = resumeAt - } catch { - // ignore - } - }, - { once: true }, - ) + onDirectMetadata = () => { + // STRM/115 直链要等 seekable 覆盖目标位置再写 currentTime（否则会被 + // 钳回 0），并顺带打开误报宽限期，避免这次跳转被当成播放失败又弹回 HLS。 + if (isStrmMedia(currentMedia)) { + seekStrmDirectTo(resumeAt) + return + } + try { + video.currentTime = resumeAt + } catch { + // ignore + } + } + video.addEventListener('loadedmetadata', onDirectMetadata, { once: true }) } video.src = url void video.play().catch(() => undefined) @@ -1457,6 +1451,8 @@ export function PlayerPage() { return () => { cancelled = true video.removeEventListener('playing', onPlaying) + if (onDirectMetadata) video.removeEventListener('loadedmetadata', onDirectMetadata) + cleanupHlsResume?.() teardownHls() } }, [ @@ -1505,19 +1501,22 @@ export function PlayerPage() { if (!id) return setResumePosition(0) setInitialSeekDone(false) + const controller = new AbortController() playbackAPI - .getResume(id) + .getResume(id, controller.signal) .then((progress) => { + if (controller.signal.aborted) return if (progress.position_ms > 2000 && !progress.completed) { setResumePosition(progress.position_ms / 1000) } }) .catch(() => undefined) + return () => controller.abort() }, [id]) useEffect(() => { if (!resumePosition || initialSeekDone) return - if (mode === 'hls') { + if (mode === 'hls' && hlsSource === 'local') { // Restart transcode near the resume point instead of seeking a short partial playlist. if (Math.abs(hlsStartSec - resumePosition) > 2) { setHlsStartSec(resumePosition) @@ -1536,6 +1535,7 @@ export function PlayerPage() { let seekStarted = false let onPlaying: (() => void) | undefined let onCanPlay: (() => void) | undefined + let onSeeked: (() => void) | undefined const notifyResume = () => { const m = Math.floor(resumePosition / 60) @@ -1552,7 +1552,7 @@ export function PlayerPage() { return } seekStarted = true - if (isStrmMedia(mediaRef.current)) { + if (mode === 'direct' && isStrmMedia(mediaRef.current)) { seekStrmDirectTo(resumePosition, { timeoutMs: 6000, onSettled: () => { @@ -1579,7 +1579,8 @@ export function PlayerPage() { directSeekGraceUntilRef.current = Date.now() + DIRECT_SEEK_GRACE_MS directRetryRef.current = false clearFallbackTimer() - const onSeeked = () => { + onSeeked = () => { + if (cancelled) return setInitialSeekDone(true) notifyResume() } @@ -1592,7 +1593,7 @@ export function PlayerPage() { } // STRM/云盘：等真正起播后再续播跳转，避免 302 未完成就 seek。 - const deferForStrm = isStrmMedia(mediaRef.current) + const deferForStrm = mode === 'direct' && isStrmMedia(mediaRef.current) if (deferForStrm) { onPlaying = () => { performResumeSeek() @@ -1615,8 +1616,9 @@ export function PlayerPage() { cancelled = true if (onPlaying) video.removeEventListener('playing', onPlaying) if (onCanPlay) video.removeEventListener('canplay', onCanPlay) + if (onSeeked) video.removeEventListener('seeked', onSeeked) } - }, [resumePosition, initialSeekDone, mode, hlsStartSec, clearFallbackTimer, seekStrmDirectTo]) + }, [resumePosition, initialSeekDone, mode, hlsSource, hlsStartSec, clearFallbackTimer, seekStrmDirectTo]) // 使用 ref 实时同步进度计算所需的状态，避免每次 hlsStartSec 改变都触发 cleanup 并误上报旧进度 const hlsStartSecRef = useRef(hlsStartSec) @@ -2223,28 +2225,22 @@ export function PlayerPage() { toast.error('图片字幕需要开启 HLS 转码后才能显示') return } + const position = playbackPositionSec(mode, hlsSource, hlsStartSec, ref.current?.currentTime || 0) if (nextTrack?.delivery === 'burn') { // PGS 等图片字幕必须走本地 FFmpeg 烧录，不能交给云端 HLS。 - switchToLocalHLS(ref.current?.currentTime || 0) - } - const burnChanged = - oldTrack?.delivery === 'burn' || nextTrack?.delivery === 'burn' - if (burnChanged && mode === 'hls' && ref.current) { - setHlsStartSec(hlsStartSec + (ref.current.currentTime || 0)) + switchToLocalHLS(position) + } else if (oldTrack?.delivery === 'burn' && mode === 'hls') { + setHlsStartSec(position) } setSubtitleIndex(index) - if (nextTrack?.delivery === 'burn' && mode !== 'hls' && !directOnly && !isDirectStream) { - setHlsStartSec(ref.current?.currentTime || 0) - setPlaybackMode('hls') - } - }, [directOnly, hlsStartSec, isDirectStream, mode, setPlaybackMode, subs, subtitleIndex, switchToLocalHLS]) + }, [directOnly, hlsSource, hlsStartSec, isDirectStream, mode, subs, subtitleIndex, switchToLocalHLS]) const selectPlaybackQuality = useCallback( (quality: PlaybackQuality) => { const video = ref.current const localTime = video?.currentTime || 0 // 云 HLS 的 currentTime 本来就在绝对时间轴上；本地 HLS 的 t=0 对应 hlsStartSec。 - const position = mode === 'hls' && hlsSource === 'local' ? hlsStartSec + localTime : localTime + const position = playbackPositionSec(mode, hlsSource, hlsStartSec, localTime) if (quality.source === 'original') { setCloudWaiting(false) setHlsSource('local') @@ -2256,12 +2252,13 @@ export function PlayerPage() { if (quality.source === 'cloud') { setSelectedQuality(quality.id) setHlsSource('cloud') + setHlsStartSec(position) + pendingSeekRef.current = position > 2 ? position : null // 用户手动选了云端档位：这是一次明确的重新尝试，重置自动重连的次数窗口。 cloudReloadsRef.current = [] if (quality.available) { setCloudWaiting(false) setCloudWaitMessage('') - if (position > 2) pendingSeekRef.current = position setPlaybackMode('hls') } else { setCloudWaiting(true) @@ -2281,7 +2278,8 @@ export function PlayerPage() { toast.error(quality.note || '当前无法进行 HLS 转码，请在「系统设置 → 常规」安装 ffmpeg 后重试') return } - if (position > 2) setHlsStartSec(position) + pendingSeekRef.current = null + setHlsStartSec(position) setPlaybackMode('hls') }, [hlsSource, hlsStartSec, mode, setPlaybackMode, startCloudTranscode], diff --git a/web/src/pages/hlsPlayback.test.ts b/web/src/pages/hlsPlayback.test.ts new file mode 100644 index 0000000..25606b0 --- /dev/null +++ b/web/src/pages/hlsPlayback.test.ts @@ -0,0 +1,58 @@ +import { strict as assert } from 'node:assert' +import { applyPendingHlsSeek, attachNativeHls, bindPendingHlsSeek, hlsManifestTimeoutMs, playbackPositionSec } from './hlsPlayback.ts' + +class FakeVideo extends EventTarget { + src = '' + rejectSeek = false + position = 0 + playCalls = 0 + get currentTime() { return this.position } + set currentTime(value: number) { + if (this.rejectSeek) throw new Error('metadata not ready') + this.position = value + } + play() { this.playCalls += 1; return Promise.resolve() } +} + +const video = new FakeVideo() +const pending = { current: 420 as number | null } +const detach = attachNativeHls(video, '/cloud/master.m3u8', pending) +assert.equal(video.src, '/cloud/master.m3u8') +assert.equal(video.playCalls, 1) +assert.equal(video.currentTime, 0, 'seek must wait for the new source metadata') +video.rejectSeek = true +video.dispatchEvent(new Event('loadedmetadata')) +assert.equal(pending.current, 420, 'a rejected seek must survive until canplay') +video.rejectSeek = false +video.dispatchEvent(new Event('canplay')) +assert.equal(video.currentTime, 420, 'native HLS must restore the absolute cloud position') +assert.equal(pending.current, null) +pending.current = 600 +detach() +video.dispatchEvent(new Event('loadedmetadata')) +assert.equal(video.currentTime, 420, 'a detached source must not apply late metadata events') +assert.equal(pending.current, 600) + +const msePending = { current: 120 as number | null } +video.rejectSeek = true +assert.equal(applyPendingHlsSeek(video, msePending), false) +assert.equal(msePending.current, 120) +video.rejectSeek = false +assert.equal(applyPendingHlsSeek(video, msePending), true) +assert.equal(msePending.current, null) +const lateMsePending = { current: 900 as number | null } +video.rejectSeek = true +const detachMse = bindPendingHlsSeek(video, lateMsePending) +assert.equal(applyPendingHlsSeek(video, lateMsePending), false) +video.rejectSeek = false +video.dispatchEvent(new Event('loadedmetadata')) +assert.equal(video.currentTime, 900, 'hls.js seeks rejected during manifest parsing must retry after metadata') +assert.equal(lateMsePending.current, null) +detachMse() +assert.ok(hlsManifestTimeoutMs('local', 1800) > 120_000, 'deep seeks must outlast the server readiness deadline') +assert.equal(hlsManifestTimeoutMs('local', 0), 60_000) +assert.equal(hlsManifestTimeoutMs('cloud', 1800), 60_000) +assert.equal(playbackPositionSec('hls', 'cloud', 420, 600), 600, 'cloud subtitle/quality switches must not add an old local offset') +assert.equal(playbackPositionSec('hls', 'local', 420, 60), 480, 'local subtitle switches preserve the source position') +assert.equal(playbackPositionSec('direct', 'local', 420, 60), 60) +console.log('hlsPlayback.test.ts ok') diff --git a/web/src/pages/hlsPlayback.ts b/web/src/pages/hlsPlayback.ts new file mode 100644 index 0000000..a73f3a8 --- /dev/null +++ b/web/src/pages/hlsPlayback.ts @@ -0,0 +1,43 @@ +type HlsVideo = Pick +type PendingSeek = { current: number | null } + +/** Local mid-file restarts may wait 120 seconds on the server before a playlist exists. */ +export function hlsManifestTimeoutMs(source: 'cloud' | 'local', startSec: number): number { + return source === 'local' && startSec > 0.05 ? 135_000 : 60_000 +} + +/** Keep a pending seek until the media element accepts it; metadata can arrive later. */ +export function applyPendingHlsSeek(video: Pick, pending: PendingSeek): boolean { + if (pending.current === null) return true + if (!Number.isFinite(pending.current) || pending.current < 0) return false + try { + video.currentTime = pending.current + pending.current = null + return true + } catch { + return false + } +} + +export function bindPendingHlsSeek(video: HlsVideo, pending: PendingSeek): () => void { + const restore = () => { applyPendingHlsSeek(video, pending) } + video.addEventListener('loadedmetadata', restore) + video.addEventListener('canplay', restore) + return () => { + video.removeEventListener('loadedmetadata', restore) + video.removeEventListener('canplay', restore) + } +} + +/** Native HLS needs the same seek restoration as hls.js, with removable listeners. */ +export function attachNativeHls(video: HlsVideo, url: string, pending: PendingSeek): () => void { + const detach = bindPendingHlsSeek(video, pending) + video.src = url + void video.play().catch(() => undefined) + return detach +} + +/** Only locally transcoded HLS uses a relative timeline starting at startSec. */ +export function playbackPositionSec(mode: 'direct' | 'hls', source: 'cloud' | 'local', startSec: number, currentTime: number): number { + return mode === 'hls' && source === 'local' ? startSec + currentTime : currentTime +}