fix media event access, cache isolation and indexing

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

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

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

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