diff --git a/internal/handler/media.go b/internal/handler/media.go
index da97844..c3e2bea 100644
--- a/internal/handler/media.go
+++ b/internal/handler/media.go
@@ -556,19 +556,17 @@ func listMediaHandler(svc *service.Container) gin.HandlerFunc {
})
return
}
- grouped, err := svc.Media.GroupedMediaVisibleFiltered(ctx, id, mediaVisibilityForRequest(c, svc), filters)
+ items, total, err := svc.Media.ListMediaVisibleSortedPage(ctx, id, page, size, mediaVisibilityForRequest(c, svc), filters, sortSpec, history)
if err != nil {
writeInternalOrCanceled(c, err)
return
}
- grouped = service.SortMediaItems(grouped, sortSpec.Field, sortSpec.Order, history)
- items := service.PaginateMediaItems(grouped, page, size)
if items == nil {
items = []service.MediaItem{}
}
c.JSON(http.StatusOK, gin.H{
"items": items,
- "total": len(grouped),
+ "total": total,
"page": page,
"page_size": size,
})
diff --git a/internal/handler/routes_authenticated_features.go b/internal/handler/routes_authenticated_features.go
index 2769c1a..e96892b 100644
--- a/internal/handler/routes_authenticated_features.go
+++ b/internal/handler/routes_authenticated_features.go
@@ -21,6 +21,8 @@ func registerAuthedDLNARoutes(authed *gin.RouterGroup, svc *service.Container) {
}
func registerAuthedRealtimeRoutes(authed *gin.RouterGroup, svc *service.Container) {
- authed.GET("/ws", wsHandler(svc))
- authed.GET("/events", sseHandler(svc))
+ // These streams carry operator tasks, including filesystem paths and scan
+ // errors. Their payloads are not scoped to an individual user's libraries.
+ authed.GET("/ws", middleware.AdminRequired(), wsHandler(svc))
+ authed.GET("/events", middleware.AdminRequired(), sseHandler(svc))
}
diff --git a/internal/handler/ws.go b/internal/handler/ws.go
index 5ba80ce..35014a9 100644
--- a/internal/handler/ws.go
+++ b/internal/handler/ws.go
@@ -1,9 +1,7 @@
// Package handler — WebSocket endpoint.
//
-// Clients connect to /api/ws?token=... (the token is the same JWT used for
-// REST calls). The first message they send is a JSON {"action":"subscribe",
-// "topics":["scan","scrape","transcode"]}. Subsequent server-pushed events
-// arrive as {"topic":"...","payload":{...}}.
+// Administrators connect to /api/ws?token=... with their REST access token.
+// Operator events arrive as {"topic":"...","payload":{...}}.
package handler
import (
@@ -52,8 +50,11 @@ func wsHandler(svc *service.Container) gin.HandlerFunc {
sub := svc.WSHub.Subscribe(id, nil)
defer svc.WSHub.Unsubscribe(id)
- // Reader: accept ping + subscription updates.
+ // Reading processes WebSocket control frames and detects disconnects.
+ done := make(chan struct{})
+ conn.SetReadLimit(64 << 10)
go func() {
+ defer close(done)
for {
if _, _, err := conn.NextReader(); err != nil {
_ = conn.Close()
@@ -67,6 +68,8 @@ func wsHandler(svc *service.Container) gin.HandlerFunc {
defer ticker.Stop()
for {
select {
+ case <-done:
+ return
case ev, ok := <-sub.Out:
if !ok {
return
diff --git a/internal/handler/ws_test.go b/internal/handler/ws_test.go
new file mode 100644
index 0000000..9537a13
--- /dev/null
+++ b/internal/handler/ws_test.go
@@ -0,0 +1,93 @@
+package handler
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/gin-gonic/gin"
+ "github.com/golang-jwt/jwt/v5"
+ "github.com/gorilla/websocket"
+ "go.uber.org/zap"
+
+ "github.com/truewhile/MeBox/internal/middleware"
+ "github.com/truewhile/MeBox/internal/service"
+)
+
+func realtimeTestRouter(hub *service.Hub) *gin.Engine {
+ gin.SetMode(gin.TestMode)
+ router := gin.New()
+ authed := router.Group("/api")
+ authed.Use(middleware.AuthRequired("realtime-test"))
+ registerAuthedRealtimeRoutes(authed, &service.Container{WSHub: hub})
+ return router
+}
+
+func realtimeTestToken(t *testing.T, role string) string {
+ t.Helper()
+ token, err := jwt.NewWithClaims(jwt.SigningMethodHS256, middleware.Claims{
+ UserID: "viewer", Role: role,
+ RegisteredClaims: jwt.RegisteredClaims{ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour))},
+ }).SignedString([]byte("realtime-test"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return token
+}
+
+func TestRealtimeStreamsRejectOrdinaryUsers(t *testing.T) {
+ router := realtimeTestRouter(nil)
+ for _, path := range []string{"/api/ws", "/api/events"} {
+ for _, role := range []string{"", "user"} {
+ req := httptest.NewRequest(http.MethodGet, path, nil)
+ want := http.StatusUnauthorized
+ if role != "" {
+ req.Header.Set("Authorization", "Bearer "+realtimeTestToken(t, role))
+ want = http.StatusForbidden
+ }
+ recorder := httptest.NewRecorder()
+ router.ServeHTTP(recorder, req)
+ if recorder.Code != want {
+ t.Fatalf("%s role %q: status %d, want %d", path, role, recorder.Code, want)
+ }
+ }
+ }
+}
+
+func TestAdminWebSocketReceivesTaskEvents(t *testing.T) {
+ hub := service.NewHub(zap.NewNop())
+ go hub.Run()
+ defer hub.Stop()
+ server := httptest.NewServer(realtimeTestRouter(hub))
+ defer server.Close()
+ url := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/ws?token=" + realtimeTestToken(t, "admin")
+ conn, _, err := websocket.DefaultDialer.Dial(url, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer conn.Close()
+ stop := make(chan struct{})
+ defer close(stop)
+ go func() {
+ ticker := time.NewTicker(10 * time.Millisecond)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-stop:
+ return
+ case <-ticker.C:
+ hub.Publish("task", service.BackgroundTask{ID: "task-1", SourcePath: "/media/private"})
+ }
+ }
+ }()
+ _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+ var event service.Event
+ if err := conn.ReadJSON(&event); err != nil {
+ t.Fatal(err)
+ }
+ if event.Topic != "task" || event.Payload.(map[string]any)["source_path"] != "/media/private" {
+ t.Fatalf("event = %#v", event)
+ }
+}
diff --git a/internal/repository/media_repository.go b/internal/repository/media_repository.go
index 091844f..37f9a67 100644
--- a/internal/repository/media_repository.go
+++ b/internal/repository/media_repository.go
@@ -178,14 +178,6 @@ func (r *MediaRepository) YearRange(ctx context.Context, filter MediaQueryFilter
return min, max, nil
}
-func (r *MediaRepository) indexMediaBestEffort(ctx context.Context, media model.Media) {
- backend, ok := r.searchBackend.(MediaSearchSyncBackend)
- if !ok {
- return
- }
- _ = backend.IndexMedia(ctx, []model.Media{media})
-}
-
// FindByID returns the media row or (nil, nil).
func (r *MediaRepository) FindByID(ctx context.Context, id string) (*model.Media, error) {
var m model.Media
diff --git a/internal/repository/media_repository_upsert.go b/internal/repository/media_repository_upsert.go
index 84b0554..9db85a1 100644
--- a/internal/repository/media_repository_upsert.go
+++ b/internal/repository/media_repository_upsert.go
@@ -96,15 +96,34 @@ 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 {
+ backend, ok := r.searchBackend.(MediaSearchSyncBackend)
+ if !ok {
return
}
+ unique := make([]string, 0, len(ids))
+ seen := make(map[string]struct{}, len(ids))
for _, id := range ids {
if id == "" {
continue
}
- if fresh, err := r.FindByID(ctx, id); err == nil && fresh != nil {
- r.indexMediaBestEffort(ctx, *fresh)
+ if _, exists := seen[id]; !exists {
+ seen[id] = struct{}{}
+ unique = append(unique, id)
+ }
+ }
+ // Read committed rows and submit one bulk per bounded batch. Keep network
+ // work outside the transaction and below SQLite's parameter limit.
+ const batchSize = 500
+ for start := 0; start < len(unique); start += batchSize {
+ batch := unique[start:min(start+batchSize, len(unique))]
+ rows, err := r.ListMediaByIDs(ctx, batch)
+ if err == nil {
+ err = backend.IndexMedia(ctx, rows)
+ }
+ if err != nil {
+ // The primary write has committed; do not report a failed upsert and
+ // make scanners repeat it. Surface the separate index failure instead.
+ r.db.Logger.Error(ctx, "media saved but search index sync failed for %d items: %v", len(batch), err)
}
}
}
diff --git a/internal/repository/opensearch.go b/internal/repository/opensearch.go
index 9b8ecdc..6d931ed 100644
--- a/internal/repository/opensearch.go
+++ b/internal/repository/opensearch.go
@@ -4,8 +4,10 @@ import (
"bytes"
"context"
"encoding/json"
+ "errors"
"fmt"
"io"
+ "net"
"net/http"
"net/url"
"strings"
@@ -135,14 +137,113 @@ func (b *OpenSearchMediaBackend) EnsureIndex(ctx context.Context) error {
}
func (b *OpenSearchMediaBackend) IndexMedia(ctx context.Context, rows []model.Media) error {
+ const batchSize = 500
+ var failures []error
+ for start := 0; start < len(rows); start += batchSize {
+ if err := b.indexMediaBatch(ctx, rows[start:min(start+batchSize, len(rows))]); err != nil {
+ failures = append(failures, err)
+ if ctx.Err() != nil {
+ break
+ }
+ }
+ }
+ return errors.Join(failures...)
+}
+
+type openSearchBulkResponse struct {
+ Errors bool `json:"errors"`
+ Items []struct {
+ Index struct {
+ ID string `json:"_id"`
+ Status int `json:"status"`
+ Error *struct {
+ Type string `json:"type"`
+ } `json:"error"`
+ } `json:"index"`
+ } `json:"items"`
+}
+
+func (b *OpenSearchMediaBackend) indexMediaBatch(ctx context.Context, rows []model.Media) error {
if len(rows) == 0 {
return nil
}
+ pending := rows
+ var failures []error
+ const attempts = 3
+ for attempt := 0; attempt < attempts; attempt++ {
+ if err := ctx.Err(); err != nil {
+ return errors.Join(append(failures, err)...)
+ }
+ if attempt > 0 {
+ timer := time.NewTimer(time.Duration(100*(1<<(attempt-1))) * time.Millisecond)
+ select {
+ case <-ctx.Done():
+ timer.Stop()
+ return errors.Join(append(failures, ctx.Err())...)
+ case <-timer.C:
+ }
+ }
+ response, err := b.submitMediaBulk(ctx, pending)
+ if err != nil {
+ if retryableOpenSearchError(err) && attempt < attempts-1 {
+ continue
+ }
+ return errors.Join(append(failures, err)...)
+ }
+ if len(response.Items) != len(pending) {
+ return errors.Join(append(failures, fmt.Errorf("opensearch bulk returned %d item results for %d documents", len(response.Items), len(pending)))...)
+ }
+ var retry []model.Media
+ failed := false
+ for i, item := range response.Items {
+ result := item.Index
+ if result.Status >= 200 && result.Status < 300 && result.Error == nil {
+ continue
+ }
+ failed = true
+ if retryableOpenSearchStatus(result.Status) && attempt < attempts-1 {
+ retry = append(retry, pending[i])
+ continue
+ }
+ errorType := "unknown"
+ if result.Error != nil {
+ errorType = result.Error.Type
+ }
+ failures = append(failures, fmt.Errorf("opensearch bulk document %q failed: status %d (%s)", pending[i].ID, result.Status, errorType))
+ }
+ if response.Errors && !failed {
+ failures = append(failures, errors.New("opensearch bulk reported errors without failed item results"))
+ }
+ if len(retry) == 0 {
+ break
+ }
+ pending = retry
+ }
+ return errors.Join(failures...)
+}
+
+func retryableOpenSearchStatus(status int) bool {
+ return status == http.StatusRequestTimeout || status == http.StatusTooManyRequests || status >= 500
+}
+
+func retryableOpenSearchError(err error) bool {
+ var statusError *openSearchHTTPError
+ if errors.As(err, &statusError) {
+ return retryableOpenSearchStatus(statusError.status)
+ }
+ var networkError net.Error
+ return errors.As(err, &networkError)
+}
+
+func (b *OpenSearchMediaBackend) submitMediaBulk(ctx context.Context, rows []model.Media) (*openSearchBulkResponse, error) {
+ if len(rows) == 0 {
+ return &openSearchBulkResponse{}, nil
+ }
var bulk bytes.Buffer
enc := json.NewEncoder(&bulk)
for _, row := range rows {
if err := enc.Encode(map[string]any{"index": map[string]any{"_index": b.index, "_id": row.ID}}); err != nil {
- return err
+ return nil, err
}
if err := enc.Encode(map[string]any{
"id": row.ID,
@@ -155,10 +256,24 @@ func (b *OpenSearchMediaBackend) IndexMedia(ctx context.Context, rows []model.Me
"deleted": row.DeletedAt.Valid,
"created_at": row.CreatedAt,
}); err != nil {
- return err
+ return nil, err
}
}
- return b.do(ctx, http.MethodPost, "/_bulk", &bulk, "application/x-ndjson", nil)
+ var response openSearchBulkResponse
+ if err := b.do(ctx, http.MethodPost, "/_bulk", &bulk, "application/x-ndjson", &response); err != nil {
+ return nil, err
+ }
+ return &response, nil
+}
+
+type openSearchHTTPError struct {
+ method string
+ path string
+ status int
+}
+
+func (e *openSearchHTTPError) Error() string {
+ return fmt.Sprintf("opensearch %s %s returned %d", e.method, e.path, e.status)
}
func (b *OpenSearchMediaBackend) doJSON(ctx context.Context, method, path string, body any, out any) error {
@@ -190,7 +305,7 @@ func (b *OpenSearchMediaBackend) do(ctx context.Context, method, path string, bo
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
- return fmt.Errorf("opensearch %s %s returned %d", method, path, resp.StatusCode)
+ return &openSearchHTTPError{method: method, path: path, status: resp.StatusCode}
}
if out == nil {
_, _ = io.Copy(io.Discard, resp.Body)
diff --git a/internal/repository/opensearch_test.go b/internal/repository/opensearch_test.go
index c3bd806..309e33c 100644
--- a/internal/repository/opensearch_test.go
+++ b/internal/repository/opensearch_test.go
@@ -1,12 +1,23 @@
package repository
import (
+ "bufio"
+ "context"
"encoding/json"
+ "errors"
+ "fmt"
"net/http"
"net/http/httptest"
+ "strings"
"testing"
+ "time"
+
+ "github.com/glebarez/sqlite"
+ "gorm.io/gorm"
+ "gorm.io/gorm/logger"
"github.com/truewhile/MeBox/internal/config"
+ "github.com/truewhile/MeBox/internal/model"
)
func TestOpenSearchMediaBackendSearchesIDs(t *testing.T) {
@@ -55,3 +66,156 @@ func TestOpenSearchMediaBackendSearchesIDs(t *testing.T) {
t.Fatalf("ids=%#v total=%d", ids, total)
}
}
+
+func bulkRequestIDs(r *http.Request) []string {
+ var ids []string
+ scanner := bufio.NewScanner(r.Body)
+ for scanner.Scan() {
+ var action struct {
+ Index struct {
+ ID string `json:"_id"`
+ } `json:"index"`
+ }
+ _ = json.Unmarshal(scanner.Bytes(), &action)
+ ids = append(ids, action.Index.ID)
+ if !scanner.Scan() {
+ break
+ }
+ }
+ return ids
+}
+
+func writeBulkResults(w http.ResponseWriter, ids []string, statuses map[string]int) {
+ items := make([]any, 0, len(ids))
+ failed := false
+ for _, id := range ids {
+ status := statuses[id]
+ if status == 0 {
+ status = http.StatusCreated
+ }
+ result := map[string]any{"_id": id, "status": status}
+ if status >= 400 {
+ failed = true
+ result["error"] = map[string]any{"type": "test_failure"}
+ }
+ items = append(items, map[string]any{"index": result})
+ }
+ _ = json.NewEncoder(w).Encode(map[string]any{"errors": failed, "items": items})
+}
+
+func TestOpenSearchBulkRetriesOnlyTransientFailedDocuments(t *testing.T) {
+ var requests [][]string
+ upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ ids := bulkRequestIDs(r)
+ requests = append(requests, ids)
+ statuses := map[string]int{}
+ if len(requests) == 1 {
+ statuses["retry"] = http.StatusTooManyRequests
+ statuses["invalid"] = http.StatusBadRequest
+ }
+ writeBulkResults(w, ids, statuses)
+ }))
+ defer upstream.Close()
+ backend := NewOpenSearchMediaBackend(config.SearchConfig{Backend: "opensearch", OpenSearchURL: upstream.URL})
+ rows := []model.Media{{Base: model.Base{ID: "ok"}}, {Base: model.Base{ID: "retry"}}, {Base: model.Base{ID: "invalid"}}}
+ err := backend.IndexMedia(t.Context(), rows)
+ if err == nil || !strings.Contains(err.Error(), `"invalid"`) {
+ t.Fatalf("error = %v, want permanent document failure", err)
+ }
+ if len(requests) != 2 || strings.Join(requests[1], ",") != "retry" {
+ t.Fatalf("requests = %#v; successful/permanent documents must not be retried", requests)
+ }
+}
+
+func TestOpenSearchBulkBoundsRetriesAndRejectsMalformedResponses(t *testing.T) {
+ for _, scenario := range []string{"transient-item", "transient-http", "invalid-http", "missing-results"} {
+ t.Run(scenario, func(t *testing.T) {
+ calls := 0
+ upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ calls++
+ switch scenario {
+ case "transient-item":
+ writeBulkResults(w, bulkRequestIDs(r), map[string]int{"m1": http.StatusServiceUnavailable})
+ case "transient-http":
+ w.WriteHeader(http.StatusServiceUnavailable)
+ case "invalid-http":
+ w.WriteHeader(http.StatusBadRequest)
+ case "missing-results":
+ fmt.Fprint(w, `{"errors":false,"items":[]}`)
+ }
+ }))
+ defer upstream.Close()
+ backend := NewOpenSearchMediaBackend(config.SearchConfig{Backend: "opensearch", OpenSearchURL: upstream.URL})
+ if err := backend.IndexMedia(t.Context(), []model.Media{{Base: model.Base{ID: "m1"}}}); err == nil {
+ t.Fatal("failed indexing was reported as successful")
+ }
+ want := 1
+ if strings.HasPrefix(scenario, "transient") {
+ want = 3
+ }
+ if calls != want {
+ t.Fatalf("calls = %d, want %d", calls, want)
+ }
+ })
+ }
+}
+
+func TestOpenSearchBulkHonorsCancellationDuringRetry(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ calls := 0
+ upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ calls++
+ writeBulkResults(w, bulkRequestIDs(r), map[string]int{"m1": http.StatusTooManyRequests})
+ cancel()
+ }))
+ defer upstream.Close()
+ backend := NewOpenSearchMediaBackend(config.SearchConfig{Backend: "opensearch", OpenSearchURL: upstream.URL})
+ err := backend.IndexMedia(ctx, []model.Media{{Base: model.Base{ID: "m1"}}})
+ if !errors.Is(err, context.Canceled) || calls != 1 {
+ t.Fatalf("error = %v, calls = %d", err, calls)
+ }
+}
+
+func TestMediaUpsertBatchIndexesCommittedRowsTogether(t *testing.T) {
+ db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ sqlDB, _ := db.DB()
+ sqlDB.SetMaxOpenConns(1)
+ defer sqlDB.Close()
+ if err := db.AutoMigrate(&model.Media{}); err != nil {
+ t.Fatal(err)
+ }
+ repos := New(db)
+ ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+ defer cancel()
+ calls := 0
+ upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ calls++
+ ids := bulkRequestIDs(r)
+ if len(ids) != 5 {
+ t.Errorf("bulk contains %d documents, want 5", len(ids))
+ }
+ // With only one DB connection this also detects network work inside the
+ // transaction: a separate read could not observe the committed batch.
+ var count int64
+ if err := db.WithContext(ctx).Model(&model.Media{}).Count(&count).Error; err != nil || count != 5 {
+ t.Errorf("committed rows = %d, error = %v", count, err)
+ }
+ writeBulkResults(w, ids, nil)
+ }))
+ defer upstream.Close()
+ repos.Media.SetSearchBackend(NewOpenSearchMediaBackend(config.SearchConfig{Backend: "opensearch", OpenSearchURL: upstream.URL}))
+ var rows []*model.Media
+ for i := 0; i < 5; i++ {
+ rows = append(rows, &model.Media{LibraryID: "lib", Path: fmt.Sprintf("/media/%d.mkv", i)})
+ }
+ if err := repos.Media.UpsertBatch(ctx, rows); err != nil {
+ t.Fatal(err)
+ }
+ if calls != 1 {
+ t.Fatalf("index requests = %d, want one bulk", calls)
+ }
+}
diff --git a/internal/service/media.go b/internal/service/media.go
index ffa89b5..b844850 100644
--- a/internal/service/media.go
+++ b/internal/service/media.go
@@ -12,13 +12,14 @@ import (
// MediaService offers high-level CRUD over libraries and media items.
type MediaService struct {
- cfg *config.Config
- log *zap.Logger
- repo *repository.Container
- cache *RuntimeCacheService
- groupedMediaFlight singleflight.Group
- libraryRowsFlight singleflight.Group
- recentSeriesFlight singleflight.Group
+ cfg *config.Config
+ log *zap.Logger
+ repo *repository.Container
+ cache *RuntimeCacheService
+ groupedMediaFlight singleflight.Group
+ mediaSortFlight singleflight.Group
+ libraryRowsFlight singleflight.Group
+ recentSeriesFlight singleflight.Group
}
type MediaVisibility struct {
diff --git a/internal/service/media_listing.go b/internal/service/media_listing.go
index 077a29f..cb20361 100644
--- a/internal/service/media_listing.go
+++ b/internal/service/media_listing.go
@@ -5,7 +5,7 @@ import (
"strings"
"time"
- "go.uber.org/zap"
+ "github.com/google/uuid"
"github.com/truewhile/MeBox/internal/model"
"github.com/truewhile/MeBox/internal/repository"
@@ -129,6 +129,26 @@ func (s *MediaService) GroupedMediaVisibleFiltered(
visibility MediaVisibility,
filters MediaListFilters,
) ([]MediaItem, error) {
+ value, err := s.groupedMediaVisibleFiltered(ctx, libraryID, visibility, filters)
+ if err != nil {
+ return nil, err
+ }
+ return value.Items, nil
+}
+
+// Each immutable grouping has its own revision. Lightweight sort indices can
+// reference that revision without retaining another copy of the full rows.
+type groupedMediaCacheValue struct {
+ Items []MediaItem
+ Revision string
+}
+
+func (s *MediaService) groupedMediaVisibleFiltered(
+ ctx context.Context,
+ libraryID string,
+ visibility MediaVisibility,
+ filters MediaListFilters,
+) (*groupedMediaCacheValue, error) {
visibility = ExpandMediaVisibilityForMergedCloudLibraries(ctx, s.repo, visibility)
libraryIDs, err := MergedLibraryIDsForLibrary(ctx, s.repo, libraryID)
if err != nil {
@@ -143,7 +163,7 @@ func (s *MediaService) GroupedMediaVisibleFiltered(
value, err, _ := s.groupedMediaFlight.Do(itemsCacheKey, func() (any, error) {
if s.cache != nil {
if cachedObj, ok := s.cache.GetObject(itemsCacheKey); ok {
- if cached, ok := cachedObj.([]MediaItem); ok {
+ if cached, ok := cachedObj.(*groupedMediaCacheValue); ok {
return cached, nil
}
}
@@ -154,8 +174,8 @@ func (s *MediaService) GroupedMediaVisibleFiltered(
if err != nil {
return nil, err
}
- grouped := groupMediaVersions(items)
- if s.cache != nil && len(grouped) > 0 {
+ grouped := &groupedMediaCacheValue{Items: groupMediaVersions(items), Revision: uuid.NewString()}
+ if s.cache != nil && len(grouped.Items) > 0 {
s.cache.SetObject(itemsCacheKey, grouped, s.mediaObjectTTL())
}
return grouped, nil
@@ -163,10 +183,10 @@ func (s *MediaService) GroupedMediaVisibleFiltered(
if err != nil {
return nil, err
}
- if grouped, ok := value.([]MediaItem); ok {
+ if grouped, ok := value.(*groupedMediaCacheValue); ok {
return grouped, nil
}
- return nil, nil
+ return &groupedMediaCacheValue{}, nil
}
func (s *MediaService) listMediaVisibleForGrouping(ctx context.Context, libraryID string, visibility MediaVisibility, filters MediaListFilters) ([]model.Media, error) {
@@ -180,24 +200,19 @@ func (s *MediaService) listMediaVisibleForGrouping(ctx context.Context, libraryI
AllowedLibraryIDs: visibility.AllowedLibraryIDs,
HiddenLibraryIDs: visibility.HiddenLibraryIDs,
})
- // 版本分组的 URL 分页发生在 Go 进程内，响应里的 total 是分组后的数量，
- // 不需要数据库再为原始行做一次 COUNT(*)。全量 COUNT 在超大媒体库上
- // 会重复扫描整个 library_id 范围，而这里只关心是否存在截断风险。
- items, err := s.repo.Media.ListByLibrariesFilteredNoCount(ctx, libraryIDs, 0, maxMediaSearchLimit, filter)
+ // Group before paginating so versions stay together. The search safety limit
+ // must not truncate a library listing or its reported grouped total.
+ items, err := s.repo.Media.ListAllByLibrariesFilteredNoCount(ctx, libraryIDs, filter)
if err != nil {
return nil, err
}
- if len(items) >= maxMediaSearchLimit && s.log != nil {
- s.log.Warn("media version grouping may be truncated by safety limit",
- zap.String("library_id", libraryID),
- zap.Int("limit", maxMediaSearchLimit))
- }
s.attachLibraryMetadata(ctx, items)
return items, nil
}
// GetMedia returns a single media row.
-func (s *MediaService) GetMedia(ctx context.Context, id string) (*model.Media, error) { media, err := s.repo.Media.FindByID(ctx, id)
+func (s *MediaService) GetMedia(ctx context.Context, id string) (*model.Media, error) {
+ media, err := s.repo.Media.FindByID(ctx, id)
if err != nil || media == nil {
return media, err
}
diff --git a/internal/service/media_sorted_page.go b/internal/service/media_sorted_page.go
new file mode 100644
index 0000000..bd98c59
--- /dev/null
+++ b/internal/service/media_sorted_page.go
@@ -0,0 +1,77 @@
+package service
+
+import (
+ "context"
+ "fmt"
+ "sort"
+ "time"
+)
+
+// ListMediaVisibleSortedPage reuses grouped rows and a small position index.
+// Only the requested page is copied; pagination does not sort the full library
+// again or retain a second full-size MediaItem slice for every sort order.
+func (s *MediaService) ListMediaVisibleSortedPage(
+ ctx context.Context,
+ libraryID string,
+ page, pageSize int,
+ visibility MediaVisibility,
+ filters MediaListFilters,
+ spec MediaSortSpec,
+ history map[string]time.Time,
+) ([]MediaItem, int64, error) {
+ grouped, err := s.groupedMediaVisibleFiltered(ctx, libraryID, visibility, filters)
+ if err != nil {
+ return nil, 0, err
+ }
+ spec = NormalizeMediaSort(spec.Field, spec.Order)
+ indices := s.sortedMediaIndices(grouped, spec, history)
+ page, pageSize = normalizeGroupedMediaPage(page, pageSize)
+ total := len(grouped.Items)
+ // Division avoids overflowing (page-1)*pageSize on an excessive page value.
+ if total == 0 || page-1 > (total-1)/pageSize {
+ return []MediaItem{}, int64(total), nil
+ }
+ start := (page - 1) * pageSize
+ end := min(start+pageSize, total)
+ items := make([]MediaItem, 0, end-start)
+ for _, index := range indices[start:end] {
+ items = append(items, grouped.Items[index])
+ }
+ return items, int64(total), nil
+}
+
+func (s *MediaService) sortedMediaIndices(grouped *groupedMediaCacheValue, spec MediaSortSpec, history map[string]time.Time) []int {
+ // Playback history changes independently of the media cache and is personal.
+ // Keep this explicit sort live rather than caching another user's history.
+ if s.cache == nil || spec.Field == "last_played" || len(grouped.Items) == 0 {
+ return sortMediaIndices(grouped.Items, spec, history)
+ }
+ key := fmt.Sprintf("media:obj:sort:%s:%s:%s", grouped.Revision, spec.Field, spec.Order)
+ value, _, _ := s.mediaSortFlight.Do(key, func() (any, error) {
+ if cached, ok := s.cache.GetObject(key); ok {
+ if indices, ok := cached.([]int); ok {
+ return indices, nil
+ }
+ }
+ indices := sortMediaIndices(grouped.Items, spec, history)
+ s.cache.SetObject(key, indices, s.mediaObjectTTL())
+ return indices, nil
+ })
+ return value.([]int)
+}
+
+func sortMediaIndices(items []MediaItem, spec MediaSortSpec, history map[string]time.Time) []int {
+ indices := make([]int, len(items))
+ for i := range indices {
+ indices[i] = i
+ }
+ sort.SliceStable(indices, func(i, j int) bool {
+ a, b := items[indices[i]], items[indices[j]]
+ cmp := compareMediaByField(a, b, spec, history)
+ if cmp == 0 {
+ cmp = compareID(a.ID, b.ID)
+ }
+ return cmp < 0
+ })
+ return indices
+}
diff --git a/internal/service/media_sorted_page_test.go b/internal/service/media_sorted_page_test.go
new file mode 100644
index 0000000..5aecde0
--- /dev/null
+++ b/internal/service/media_sorted_page_test.go
@@ -0,0 +1,141 @@
+package service
+
+import (
+ "fmt"
+ "math"
+ "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 TestSortedMediaPageIncludesRowsBeyondSearchLimit(t *testing.T) {
+ db := newServiceTestDB(t)
+ repos := repository.New(db)
+ lib := model.Library{Name: "Large", Path: "/media/large", Type: "movie", Enabled: true}
+ if err := repos.Library.Create(t.Context(), &lib); err != nil {
+ t.Fatal(err)
+ }
+ // This must cross the historical 50,000-row boundary: a smaller fixture
+ // cannot detect truncation before grouping, counting, and sorting.
+ now := time.Now()
+ for start := 0; start <= maxMediaSearchLimit; start += 500 {
+ var rows []model.Media
+ for i := start; i < min(start+500, maxMediaSearchLimit+1); i++ {
+ rows = append(rows, model.Media{
+ Base: model.Base{ID: fmt.Sprintf("large-%05d", i), CreatedAt: now, UpdatedAt: now},
+ LibraryID: lib.ID, Path: fmt.Sprintf("/media/large/%05d.mkv", i),
+ Title: fmt.Sprintf("Movie %05d", i), TMDbID: i + 1,
+ })
+ }
+ if err := db.CreateInBatches(&rows, 100).Error; err != nil {
+ t.Fatal(err)
+ }
+ }
+ svc := NewMediaService(&config.Config{}, zap.NewNop(), repos)
+ items, total, err := svc.ListMediaVisibleSortedPage(t.Context(), lib.ID, maxMediaSearchLimit/100+1, 100,
+ MediaVisibility{IncludeNSFW: true}, MediaListFilters{}, MediaSortSpec{Field: "title", Order: "asc"}, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if total != int64(maxMediaSearchLimit+1) || len(items) != 1 || items[0].ID != "large-50000" {
+ t.Fatalf("total=%d, page length=%d, want 50001 rows and final item large-50000", total, len(items))
+ }
+}
+
+func TestSortedMediaPagesReuseIndicesAndRefreshAfterInvalidation(t *testing.T) {
+ db := newServiceTestDB(t)
+ repos := repository.New(db)
+ lib := model.Library{Name: "Movies", Path: "/media/movies", Type: "movie", Enabled: true}
+ if err := repos.Library.Create(t.Context(), &lib); err != nil {
+ t.Fatal(err)
+ }
+ rows := []model.Media{
+ {Base: model.Base{ID: "alpha"}, LibraryID: lib.ID, Path: "/media/movies/a.mkv", Title: "Alpha", TMDbID: 1, SizeBytes: 100},
+ {Base: model.Base{ID: "alpha-alt"}, LibraryID: lib.ID, Path: "/media/movies/a-alt.mkv", Title: "Alpha", TMDbID: 1, SizeBytes: 50},
+ {Base: model.Base{ID: "beta"}, LibraryID: lib.ID, Path: "/media/movies/b.mkv", Title: "Beta", TMDbID: 2},
+ {Base: model.Base{ID: "hidden"}, LibraryID: lib.ID, Path: "/media/movies/hidden.mkv", Title: "Hidden", TMDbID: 3, NSFW: true},
+ }
+ if err := db.Create(&rows).Error; err != nil {
+ t.Fatal(err)
+ }
+ svc := NewMediaService(&config.Config{}, zap.NewNop(), repos).
+ SetRuntimeCache(NewRuntimeCacheService(&config.Config{}, zap.NewNop()))
+ visibility := MediaVisibility{IncludeNSFW: false}
+ spec := MediaSortSpec{Field: "title", Order: "asc"}
+ first, total, err := svc.ListMediaVisibleSortedPage(t.Context(), lib.ID, 1, 1, visibility, MediaListFilters{}, spec, nil)
+ if err != nil || total != 2 || len(first) != 1 || first[0].ID != "alpha" || len(first[0].Versions) != 2 {
+ t.Fatalf("first=%#v, total=%d, error=%v", first, total, err)
+ }
+ grouped, err := svc.groupedMediaVisibleFiltered(t.Context(), lib.ID, visibility, MediaListFilters{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ index := svc.sortedMediaIndices(grouped, spec, nil)
+ second, _, err := svc.ListMediaVisibleSortedPage(t.Context(), lib.ID, 2, 1, visibility, MediaListFilters{}, spec, nil)
+ if err != nil || len(second) != 1 || second[0].ID != "beta" {
+ t.Fatalf("second=%#v, error=%v", second, err)
+ }
+ cachedIndex := svc.sortedMediaIndices(grouped, spec, nil)
+ if &index[0] != &cachedIndex[0] {
+ t.Fatal("pagination rebuilt the sorted position index")
+ }
+ // The returned page is independently writable without corrupting cached rows.
+ first[0].Title = "Modified by caller"
+ if err := db.Model(&model.Media{}).Where("id = ?", "beta").Update("title", "Aardvark").Error; err != nil {
+ t.Fatal(err)
+ }
+ svc.invalidateMediaCache(t.Context())
+ fresh, _, err := svc.ListMediaVisibleSortedPage(t.Context(), lib.ID, 1, 1, visibility, MediaListFilters{}, spec, nil)
+ if err != nil || len(fresh) != 1 || fresh[0].ID != "beta" {
+ t.Fatalf("updated page=%#v, error=%v", fresh, err)
+ }
+ empty, _, err := svc.ListMediaVisibleSortedPage(t.Context(), lib.ID, math.MaxInt, 100, visibility, MediaListFilters{}, spec, nil)
+ if err != nil || empty == nil || len(empty) != 0 {
+ t.Fatalf("out-of-range page=%#v, error=%v", empty, err)
+ }
+}
+
+func TestMediaSortIndexDoesNotCachePersonalPlaybackHistory(t *testing.T) {
+ svc := NewMediaService(&config.Config{}, zap.NewNop(), nil).
+ SetRuntimeCache(NewRuntimeCacheService(&config.Config{}, zap.NewNop()))
+ grouped := &groupedMediaCacheValue{Revision: "history-test", Items: []MediaItem{
+ {Media: model.Media{Base: model.Base{ID: "a"}}},
+ {Media: model.Media{Base: model.Base{ID: "b"}}},
+ }}
+ spec := MediaSortSpec{Field: "last_played", Order: "desc"}
+ first := svc.sortedMediaIndices(grouped, spec, map[string]time.Time{"a": time.Now()})
+ second := svc.sortedMediaIndices(grouped, spec, map[string]time.Time{"b": time.Now()})
+ if first[0] != 0 || second[0] != 1 {
+ t.Fatalf("history sorts = %v, %v", first, second)
+ }
+}
+
+func BenchmarkMediaSortIndex(b *testing.B) {
+ items := make([]MediaItem, 10000)
+ for i := range items {
+ items[i].ID = fmt.Sprint(i)
+ items[i].Title = fmt.Sprintf("Movie %05d", len(items)-i)
+ }
+ spec := MediaSortSpec{Field: "title", Order: "asc"}
+ b.Run("sort-every-page", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ SortMediaItems(items, spec.Field, spec.Order, nil)
+ }
+ })
+ b.Run("cached-position-index", func(b *testing.B) {
+ svc := NewMediaService(&config.Config{}, zap.NewNop(), nil).
+ SetRuntimeCache(NewRuntimeCacheService(&config.Config{}, zap.NewNop()))
+ grouped := &groupedMediaCacheValue{Items: items, Revision: "benchmark"}
+ svc.sortedMediaIndices(grouped, spec, nil)
+ b.ReportAllocs()
+ for b.Loop() {
+ svc.sortedMediaIndices(grouped, spec, nil)
+ }
+ })
+}
diff --git a/web/src/components/RequireAuth.tsx b/web/src/components/RequireAuth.tsx
index 8880828..24671bb 100644
--- a/web/src/components/RequireAuth.tsx
+++ b/web/src/components/RequireAuth.tsx
@@ -1,13 +1,20 @@
-import { useEffect, type ReactNode } from 'react'
+import { Fragment, useEffect, type ReactNode } from 'react'
import { Navigate, useLocation } from 'react-router-dom'
import { useAuthStore } from '../stores/auth'
+import { usePlayProfileStore } from '../stores/playProfile'
+import { libraryCacheScope } from '../utils/libraryCache'
import { hydrateReaderSettings, startReaderSettingsSync } from '../utils/readerSettingsSync'
// Route guard: redirects to /login when no token is present.
export function RequireAuth({ children }: { children: ReactNode }) {
const token = useAuthStore((s) => s.token)
const userId = useAuthStore((s) => s.user?.id ?? null)
+ // Drop local page state and in-flight callbacks when visibility changes.
+ useAuthStore((s) => s.user)
+ usePlayProfileStore((s) => s.activeProfileId)
+ usePlayProfileStore((s) => s.activeProfilePinToken)
+ const visibilityKey = libraryCacheScope()
const location = useLocation()
// 阅读器偏好按账号存在服务端：一进入受保护区域就开启回写订阅，并在
@@ -23,7 +30,7 @@ export function RequireAuth({ children }: { children: ReactNode }) {
if (!token) {
return
}
- return <>{children}
+ return {children}
}
// Route guard: only allows users with role === "admin".
diff --git a/web/src/hooks/useWebSocket.ts b/web/src/hooks/useWebSocket.ts
index c9e202d..4fbc04a 100644
--- a/web/src/hooks/useWebSocket.ts
+++ b/web/src/hooks/useWebSocket.ts
@@ -15,6 +15,7 @@ const SLOW_RECONNECT_INTERVAL = 60_000
export function useWebSocket(onEvent: (topic: string, payload: unknown) => void) {
const ref = useRef(null)
const token = useAuthStore((s) => s.token)
+ const isAdmin = useAuthStore((s) => s.user?.role === 'admin')
const onEventRef = useRef(onEvent)
useEffect(() => {
@@ -22,7 +23,7 @@ export function useWebSocket(onEvent: (topic: string, payload: unknown) => void)
}, [onEvent])
useEffect(() => {
- if (!token) return
+ if (!token || !isAdmin) return
let closed = false
let timer: number | undefined
let reconnectAttempts = 0
@@ -88,5 +89,5 @@ export function useWebSocket(onEvent: (topic: string, payload: unknown) => void)
if (timer) window.clearTimeout(timer)
ref.current?.close()
}
- }, [token])
+ }, [token, isAdmin])
}
diff --git a/web/src/pages/useLibraryData.ts b/web/src/pages/useLibraryData.ts
index 5819dad..eb60cf7 100644
--- a/web/src/pages/useLibraryData.ts
+++ b/web/src/pages/useLibraryData.ts
@@ -158,11 +158,9 @@ export function useLibraryData(
try {
const resolved = await resolveLibrary(libraryID)
if (cancelled || seq !== requestSeqRef.current) return
- if (!lib) {
- lib = resolved
- libraryRef.current = lib
- setLibrary(lib)
- }
+ lib = resolved
+ libraryRef.current = lib
+ setLibrary(lib)
} catch {
if (!cancelled && seq === requestSeqRef.current) {
setLoading(false)
diff --git a/web/src/utils/libraryCache.ts b/web/src/utils/libraryCache.ts
index de3649c..b5a1fc4 100644
--- a/web/src/utils/libraryCache.ts
+++ b/web/src/utils/libraryCache.ts
@@ -1,89 +1,35 @@
import { libraryAPI } from '../api/library'
-import type { Library } from '../types'
+import { useAuthStore } from '../stores/auth'
+import { usePlayProfileStore } from '../stores/playProfile'
+import { createLibraryCache } from './libraryCacheState'
-// 会话级媒体库元数据缓存（不落 localStorage）。
-// 首页 / 媒体库页 / 媒体库详情页都要先拿库列表或库信息才能渲染，
-// 同一会话内反复进出这些页面时直接命中缓存，避免每次都等一整轮请求。
-// TTL 内直接返回缓存；过期后重新请求并刷新。并发调用共享同一个在途请求。
-
-const LIST_TTL_MS = 30_000
-const LIBRARY_TTL_MS = 30_000
-
-let listData: Library[] | null = null
-let listAt = 0
-let listInflight: Promise | null = null
-
-const byId = new Map()
-const byIdInflight = new Map>()
-
-function fresh(at: number, ttl: number): boolean {
- return Date.now() - at < ttl
+export function libraryCacheScope(): string {
+ const { token, user } = useAuthStore.getState()
+ const profile = usePlayProfileStore.getState()
+ return JSON.stringify([
+ Boolean(token),
+ user?.id,
+ user?.role,
+ user?.hide_adult,
+ user?.allowed_library_ids,
+ profile.activeProfileId,
+ profile.activeProfilePinToken,
+ ])
}
-function rememberList(data: Library[]) {
- listData = data
- listAt = Date.now()
- for (const lib of data) {
- byId.set(lib.id, { at: listAt, data: lib })
- }
-}
+const cache = createLibraryCache({
+ list: () => libraryAPI.list(),
+ get: (id) => libraryAPI.get(id),
+ readScope: libraryCacheScope,
+})
-// fetchLibraries 拉取库列表：TTL 内命中缓存立即返回（含 peek 先行渲染），
-// 过期或未缓存时发起请求，并发调用共享在途请求。
-export function fetchLibraries(): Promise {
- if (listData && fresh(listAt, LIST_TTL_MS)) return Promise.resolve(listData)
- if (!listInflight) {
- listInflight = libraryAPI
- .list()
- .then((rows) => {
- const data = Array.isArray(rows) ? rows : []
- rememberList(data)
- return data
- })
- .finally(() => {
- listInflight = null
- })
- }
- return listInflight
-}
+// Invalidate synchronously, including logout/login between two reads and auth
+// rehydration from another tab. Refreshing a token alone preserves the cache.
+useAuthStore.subscribe(() => cache.syncScope())
+usePlayProfileStore.subscribe(() => cache.syncScope())
-// peekLibraries 同步返回当前缓存的库列表（可能过期），没有则 null。
-// 页面用它先行渲染上一次的数据，再由 fetchLibraries 在后台补一次刷新。
-export function peekLibraries(): Library[] | null {
- return listData
-}
-
-// peekLibrary 同步返回单库缓存（库列表缓存也会回填单库索引）。
-export function peekLibrary(id: string): Library | null {
- const entry = byId.get(id)
- return entry?.data ?? null
-}
-
-// resolveLibrary 拿单个库信息：命中缓存直接返回；未命中走 /libraries/:id
-// 并回填缓存。远程 Emby 挂载库的详情接口同样返回 is_remote_emby，缓存可用。
-export function resolveLibrary(id: string): Promise {
- const entry = byId.get(id)
- if (entry && fresh(entry.at, LIBRARY_TTL_MS)) return Promise.resolve(entry.data)
- const inflight = byIdInflight.get(id)
- if (inflight) return inflight
- const request = libraryAPI
- .get(id)
- .then((lib) => {
- byId.set(id, { at: Date.now(), data: lib })
- return lib
- })
- .finally(() => {
- byIdInflight.delete(id)
- })
- byIdInflight.set(id, request)
- return request
-}
-
-// invalidate 清空缓存。管理页对库做过增删改后调用，确保下次拉到新数据。
-export function invalidateLibraries() {
- listData = null
- listAt = 0
- listInflight = null
- byId.clear()
- byIdInflight.clear()
-}
+export const fetchLibraries = cache.fetchLibraries
+export const peekLibraries = cache.peekLibraries
+export const peekLibrary = cache.peekLibrary
+export const resolveLibrary = cache.resolveLibrary
+export const invalidateLibraries = cache.invalidate
diff --git a/web/src/utils/libraryCacheState.test.ts b/web/src/utils/libraryCacheState.test.ts
new file mode 100644
index 0000000..f3dd14d
--- /dev/null
+++ b/web/src/utils/libraryCacheState.test.ts
@@ -0,0 +1,107 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+import { createLibraryCache } from './libraryCacheState.ts'
+import type { Library } from '../types/library.ts'
+
+const library = (id: string) => ({ id, name: id } as Library)
+
+function deferred() {
+ let resolve!: (value: T) => void
+ const promise = new Promise((done) => { resolve = done })
+ return { promise, resolve }
+}
+
+test('list requests share a flight and populate the detail cache', async () => {
+ const request = deferred()
+ let calls = 0
+ const cache = createLibraryCache({
+ readScope: () => 'account-a:profile-a',
+ list: () => { calls++; return request.promise },
+ get: async () => { throw new Error('detail should be cached') },
+ })
+ const first = cache.fetchLibraries()
+ assert.equal(first, cache.fetchLibraries())
+ request.resolve([library('a')])
+ await first
+ assert.equal((await cache.resolveLibrary('a')).id, 'a')
+ assert.equal((await cache.fetchLibraries())[0].id, 'a')
+ assert.equal(calls, 1)
+})
+
+test('account and profile changes discard both list and detail data', async () => {
+ let scope = 'account-a:profile-a'
+ const cache = createLibraryCache({
+ readScope: () => scope,
+ list: async () => [library(scope)],
+ get: async () => library(scope),
+ })
+ await cache.fetchLibraries()
+ scope = 'account-b:profile-a'
+ assert.equal(cache.peekLibraries(), null)
+ assert.equal(cache.peekLibrary('account-a:profile-a'), null)
+ assert.equal((await cache.fetchLibraries())[0].id, scope)
+ scope = 'account-b:profile-b'
+ assert.equal(cache.peekLibraries(), null)
+ assert.equal((await cache.resolveLibrary('shared-id')).id, scope)
+})
+
+test('invalidation rejects old list results without clearing the new flight', async () => {
+ const old = deferred()
+ const fresh = deferred()
+ let calls = 0
+ const cache = createLibraryCache({
+ readScope: () => 'a',
+ list: () => ++calls === 1 ? old.promise : fresh.promise,
+ get: async (id) => library(id),
+ })
+ const oldRequest = cache.fetchLibraries()
+ const rejected = assert.rejects(oldRequest, { name: 'AbortError' })
+ cache.invalidate()
+ const current = cache.fetchLibraries()
+ old.resolve([library('deleted')])
+ await rejected
+ assert.equal(cache.peekLibraries(), null)
+ assert.equal(cache.fetchLibraries(), current)
+ fresh.resolve([library('current')])
+ await current
+ assert.equal(cache.peekLibraries()?.[0].id, 'current')
+ assert.equal(calls, 2)
+})
+
+test('invalidation rejects old detail results without clearing the new flight', async () => {
+ const old = deferred()
+ const fresh = deferred()
+ let calls = 0
+ const cache = createLibraryCache({
+ readScope: () => 'a',
+ list: async () => [],
+ get: () => ++calls === 1 ? old.promise : fresh.promise,
+ })
+ const oldRequest = cache.resolveLibrary('same-id')
+ const rejected = assert.rejects(oldRequest, { name: 'AbortError' })
+ cache.invalidate()
+ const current = cache.resolveLibrary('same-id')
+ old.resolve(library('old'))
+ await rejected
+ assert.equal(cache.peekLibrary('same-id'), null)
+ assert.equal(cache.resolveLibrary('same-id'), current)
+ fresh.resolve(library('current'))
+ await current
+ assert.equal(cache.peekLibrary('same-id')?.id, 'current')
+})
+
+test('logout followed by login to the same account invalidates old requests', async () => {
+ let scope = 'account-a'
+ const request = deferred()
+ const cache = createLibraryCache({ readScope: () => scope, list: () => request.promise, get: async libraryID => library(libraryID) })
+ const pending = cache.fetchLibraries()
+ const rejected = assert.rejects(pending, { name: 'AbortError' })
+ scope = 'logged-out'
+ cache.syncScope()
+ scope = 'account-a'
+ cache.syncScope()
+ request.resolve([library('old-session')])
+ await rejected
+ assert.equal(cache.peekLibraries(), null)
+})
diff --git a/web/src/utils/libraryCacheState.ts b/web/src/utils/libraryCacheState.ts
new file mode 100644
index 0000000..19c3f34
--- /dev/null
+++ b/web/src/utils/libraryCacheState.ts
@@ -0,0 +1,106 @@
+import type { Library } from '../types/library.ts'
+
+const TTL_MS = 30_000
+
+// Kept independent of the API and auth stores so races can be tested with
+// deferred requests. Only the active visibility scope is retained in memory.
+export function createLibraryCache({
+ list,
+ get,
+ readScope,
+}: {
+ list: () => Promise
+ get: (id: string) => Promise
+ readScope: () => string
+}) {
+ let scope = readScope()
+ let generation = 0
+ let listData: Library[] | null = null
+ let listAt = 0
+ let listInflight: Promise | null = null
+ const byId = new Map()
+ const byIdInflight = new Map>()
+
+ function invalidate() {
+ generation += 1
+ listData = null
+ listAt = 0
+ listInflight = null
+ byId.clear()
+ byIdInflight.clear()
+ }
+
+ function syncScope() {
+ const next = readScope()
+ if (scope !== next) {
+ scope = next
+ invalidate()
+ }
+ }
+
+ function checkGeneration(expected: number) {
+ syncScope()
+ if (generation !== expected) {
+ const error = new Error('Library request belongs to an invalidated cache')
+ error.name = 'AbortError'
+ throw error
+ }
+ }
+
+ function fetchLibraries(): Promise {
+ syncScope()
+ if (listData && Date.now() - listAt < TTL_MS) return Promise.resolve(listData)
+ if (listInflight) return listInflight
+ const expected = generation
+ const request = list()
+ .then((rows) => {
+ checkGeneration(expected)
+ const data = Array.isArray(rows) ? rows : []
+ listData = data
+ listAt = Date.now()
+ byId.clear()
+ for (const lib of data) byId.set(lib.id, { at: listAt, data: lib })
+ return data
+ })
+ .finally(() => {
+ if (listInflight === request) listInflight = null
+ })
+ listInflight = request
+ return request
+ }
+
+ function resolveLibrary(id: string): Promise {
+ syncScope()
+ const entry = byId.get(id)
+ if (entry && Date.now() - entry.at < TTL_MS) return Promise.resolve(entry.data)
+ const inflight = byIdInflight.get(id)
+ if (inflight) return inflight
+ const expected = generation
+ const request = get(id)
+ .then((lib) => {
+ checkGeneration(expected)
+ byId.set(id, { at: Date.now(), data: lib })
+ return lib
+ })
+ .finally(() => {
+ if (byIdInflight.get(id) === request) byIdInflight.delete(id)
+ })
+ byIdInflight.set(id, request)
+ return request
+ }
+
+ return {
+ fetchLibraries,
+ resolveLibrary,
+ invalidate,
+ syncScope,
+ peekLibraries: () => {
+ syncScope()
+ return listData
+ },
+ peekLibrary: (id: string) => {
+ syncScope()
+ return byId.get(id)?.data ?? null
+ },
+ }
+}