feat(playback): stream kill switch + async over-cap enforcer

Add the enforcement layer on top of server-observed monitoring: a revocation
kill switch that stops any stream within ~120s and keeps it dead, plus an async
over-cap enforcer that drives kills off the live monitoring picture — entirely
off the per-segment hot path and with no client-protocol change.

- internal/streamrevoke: the central kill list. IsRevoked is a pure in-memory
  lookup safe on the request hot path; a Redis pub/sub + poll mirror keeps edge
  caches current, and a Postgres durable mirror lets kills survive a server
  restart AND a Redis flush so a restart-resilient stream cannot be reconstructed
  and re-served after being killed. A user revocation is a cutoff (kills tokens
  minted before it, spares post-reauth tokens), not a 24h ban.
- internal/streamenforcer: async over-cap brain — reads the monitoring snapshot
  and per-user limits, selects victims, and collapses every reason (exceeded
  limit, admin terminate, abuse) to the same action: write a revocation.
- Edge + native + jellycompat enforcement: proxy refuses revoked sessions on
  every request and cuts long direct-play/remux pours mid-stream; the transcode
  node guards both serve and the reconstruct path so a killed session is never
  re-spawned after a node restart; jellycompat serve surfaces close their
  kill-switch coverage holes.
- streamtoken.IssuedTime exposes the token iat the user-kill cutoff compares
  against; token IssuedTime + revocation guards wire through router, downloads,
  and admin terminate-by-id (with admin-list dedupe).
- Restore sendfile zero-copy on direct-play/remux byte counting so the monitor's
  served-byte accounting does not cost the sendfile fast path.
- migrations/sql: stream_revocations durable table.

Part of the stream monitoring & kill-switch epic.

internal/api/handlers/admin_playback_control.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"io"
+	"log/slog"
 	"net/http"
 	"time"
 
@@ -12,6 +13,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/Silo-Server/silo-server/internal/playback"
+	"github.com/Silo-Server/silo-server/internal/streamrevoke"
 )
 
 const (
@@ -20,7 +22,17 @@ const (
 )
 
 type AdminPlaybackControlHandler struct {
-	playback *PlaybackHandler
+	playback   *PlaybackHandler
+	revocation *streamrevoke.Store
+}
+
+// SetRevocationStore wires the kill switch so an admin terminate revokes the
+// stream credential (stops it at the edge and refuses reconnects), not just
+// dispatches a cooperative realtime command. Optional; nil keeps prior behavior.
+func (h *AdminPlaybackControlHandler) SetRevocationStore(store *streamrevoke.Store) {
+	if h != nil {
+		h.revocation = store
+	}
 }
 
 type playbackControlRequest struct {
@@ -145,9 +157,38 @@ func (h *AdminPlaybackControlHandler) handleSessionCommand(w http.ResponseWriter
 		return
 	}
 
+	var req playbackControlRequest
+	if err := decodeOptionalJSONBody(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
+		return
+	}
+
+	// A terminate must stick even if the client ignores the realtime command:
+	// revoke the stream credential so the edge refuses further segments and any
+	// reconnect within one propagation/poll interval. Stop stays cooperative.
+	// The revocation is written BEFORE the local session lookup: the kill needs
+	// only the id, and the stream may exist only as an edge Redis record — e.g.
+	// after a central restart, or when a client that withholds progress let the
+	// in-memory session be reaped — which is precisely the stream an operator
+	// most needs to kill. Revoking an unknown id is harmless (idempotent, TTL'd).
+	if name == playback.CommandTerminate && h.revocation != nil {
+		if err := h.revocation.RevokeSession(r.Context(), sessionID, "admin_terminate"); err != nil {
+			slog.Warn("admin terminate: revoke stream failed", "session_id", sessionID, "error", err)
+		}
+	}
+
 	session, err := h.playback.sessionMgr.GetSession(sessionID)
 	if err != nil {
 		if errors.Is(err, playback.ErrSessionNotFound) {
+			// The revocation above still cut the stream at every serve surface;
+			// only the cooperative realtime command has no local session to go to.
+			if name == playback.CommandTerminate && h.revocation != nil {
+				writeJSON(w, http.StatusAccepted, playbackControlResponse{
+					CommandID: uuid.NewString(),
+					Status:    "revoked",
+				})
+				return
+			}
 			writeError(w, http.StatusNotFound, "not_found", "Playback session not found")
 			return
 		}
@@ -155,12 +196,6 @@ func (h *AdminPlaybackControlHandler) handleSessionCommand(w http.ResponseWriter
 		return
 	}
 
-	var req playbackControlRequest
-	if err := decodeOptionalJSONBody(r, &req); err != nil {
-		writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
-		return
-	}
-
 	if requiresLivePlaybackControl(name) && (session == nil || !session.HasRealtimeConnection) {
 		writeError(w, http.StatusConflict, "realtime_unavailable", "Realtime connection unavailable for playback session")
 		return

internal/api/handlers/downloads.go

@@ -7,6 +7,7 @@ import (
 	"log/slog"
 	"net/http"
 	"strconv"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 
@@ -15,6 +16,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/downloads"
 	"github.com/Silo-Server/silo-server/internal/httpstream"
 	"github.com/Silo-Server/silo-server/internal/playback"
+	"github.com/Silo-Server/silo-server/internal/streamrevoke"
 )
 
 // DownloadService is the interface that the download handler depends on. A
@@ -44,7 +46,8 @@ type DownloadService interface {
 
 // DownloadHandler handles download endpoints.
 type DownloadHandler struct {
-	svc DownloadService
+	svc        DownloadService
+	revocation *streamrevoke.Store
 }
 
 // NewDownloadHandler creates a new DownloadHandler.
@@ -52,6 +55,28 @@ func NewDownloadHandler(svc DownloadService) *DownloadHandler {
 	return &DownloadHandler{svc: svc}
 }
 
+// SetRevocationStore wires the stream kill switch so a per-user revocation cuts
+// an IN-FLIGHT download pour. Downloads stay exempt from the live-stream cap
+// (they have their own quota), but a user whose sessions were just revoked must
+// not keep pulling a multi-GB file on a pre-revocation connection. Optional;
+// nil keeps prior behavior.
+func (h *DownloadHandler) SetRevocationStore(store *streamrevoke.Store) {
+	if h != nil {
+		h.revocation = store
+	}
+}
+
+// cutDownloadOnUserRevocation arms the shared in-flight cut for a download pour.
+// Sessionless (downloads have no stream session), so only user-kind kills apply;
+// the request entry time predates any future revocation, which is exactly the
+// user-kill cutoff contract. Returns a stop func the caller must defer.
+func (h *DownloadHandler) cutDownloadOnUserRevocation(w http.ResponseWriter, userID int) func() {
+	if h.revocation == nil {
+		return func() {}
+	}
+	return h.revocation.WatchAndCut(w, "", userID, time.Now())
+}
+
 // downloadRequest represents the JSON body for POST /downloads.
 type downloadRequest struct {
 	ContentID string        `json:"content_id"`
@@ -381,6 +406,13 @@ func (h *DownloadHandler) HandlePatchDownload(w http.ResponseWriter, r *http.Req
 }
 
 // HandleDownloadFile handles GET /downloads/{id}/file.
+//
+// This is an offline download, NOT a live stream: it creates no SessionManager
+// session and no monitor record, so it is intentionally EXEMPT from the
+// concurrent-stream cap and invisible to streammonitor. Downloads are bounded by
+// the separate download concurrency/period quota (see the download service's
+// ErrConcurrentLimitReached / ErrPeriodLimitReached). See the coverage matrix
+// "downloads" note before making downloads count against the live-stream cap.
 func (h *DownloadHandler) HandleDownloadFile(w http.ResponseWriter, r *http.Request) {
 	userID := apimw.GetUserID(r.Context())
 	if userID == 0 {
@@ -397,6 +429,10 @@ func (h *DownloadHandler) HandleDownloadFile(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
+	// In-flight kill switch: a user revocation cuts this pour mid-transfer.
+	stop := h.cutDownloadOnUserRevocation(w, userID)
+	defer stop()
+
 	profileID, deviceID, _, _ := managedIdentity(r)
 	filter := requestAccessFilter(r)
 	// Full media downloads outlive the server's absolute WriteTimeout; roll
@@ -445,6 +481,10 @@ func (h *DownloadHandler) HandleDirectDownload(w http.ResponseWriter, r *http.Re
 		return
 	}
 
+	// In-flight kill switch: a user revocation cuts this pour mid-transfer.
+	stop := h.cutDownloadOnUserRevocation(w, userID)
+	defer stop()
+
 	filter := requestAccessFilter(r)
 	if err := h.svc.ServeDirect(r.Context(), w, r, userID, fileID, r.URL.Query().Get("format"), filter); err != nil {
 		h.writeDownloadError(w, err)

internal/api/middleware/metrics.go

@@ -83,8 +83,9 @@ func (w *statusWriter) Flush() {
 	}
 }
 
-// Unwrap lets http.ResponseController reach the underlying connection (e.g.
-// for the per-response write deadlines used by streaming handlers).
+// Unwrap exposes the wrapped ResponseWriter so http.ResponseController (used by
+// the stream kill switch's in-flight cut via SetWriteDeadline) can reach the
+// underlying socket instead of stopping at this wrapper and no-oping.
 func (w *statusWriter) Unwrap() http.ResponseWriter {
 	return w.ResponseWriter
 }

internal/api/middleware/request_logger.go

@@ -114,8 +114,9 @@ func (w *requestStatusWriter) Flush() {
 	}
 }
 
-// Unwrap lets http.ResponseController reach the underlying connection (e.g.
-// for the per-response write deadlines used by streaming handlers).
+// Unwrap exposes the wrapped ResponseWriter so http.ResponseController (used by
+// the stream kill switch's in-flight cut via SetWriteDeadline) can reach the
+// underlying socket instead of stopping at this wrapper and no-oping.
 func (w *requestStatusWriter) Unwrap() http.ResponseWriter {
 	return w.ResponseWriter
 }

internal/api/router.go

@@ -64,6 +64,8 @@ import (
 	"github.com/Silo-Server/silo-server/internal/scanqueue"
 	"github.com/Silo-Server/silo-server/internal/secret"
 	"github.com/Silo-Server/silo-server/internal/sections"
+	"github.com/Silo-Server/silo-server/internal/streamrevoke"
+	"github.com/Silo-Server/silo-server/internal/streamtoken"
 	"github.com/Silo-Server/silo-server/internal/subtitles"
 	subtitleai "github.com/Silo-Server/silo-server/internal/subtitles/ai"
 	"github.com/Silo-Server/silo-server/internal/subtitles/opensubtitles"
@@ -167,6 +169,9 @@ type Dependencies struct {
 	ChapterThumbnailQueuer catalog.ChapterThumbnailQueuer
 	PlaybackRealtimeHub    *playback.RealtimeHub
 	OnUserSessionsRevoked  func(ctx context.Context, userID int)
+	// RevocationStore is the stream kill switch (may be nil). Admin terminate
+	// writes to it and integrated-mode serve paths consult it.
+	RevocationStore        *streamrevoke.Store
 	OnServerSettingUpdated func(ctx context.Context, key, value string)
 	RequestServerRestart   func(ctx context.Context) error
 	ServerRestartStatus    *handlers.ServerRestartStatusTracker
@@ -1002,6 +1007,7 @@ func NewRouter(deps Dependencies) chi.Router {
 		playbackHandler.MarkerUpdateNotifier = playback.NewMarkerUpdateNotifier(deps.SessionMgr, realtimeHub)
 		subtitleAINotifier = playback.NewSubtitleReadyNotifier(deps.SessionMgr, realtimeHub)
 		adminPlaybackControlHandler = handlers.NewAdminPlaybackControlHandler(playbackHandler)
+		adminPlaybackControlHandler.SetRevocationStore(deps.RevocationStore)
 
 		if deps.DB != nil && deps.FileRepo != nil && viewerResolver != nil && deps.Config != nil && detailSvc != nil {
 			roomTokenService := watchtogether.NewRoomTokenService(deps.Config.Auth.JWTSecret, 24*time.Hour)
@@ -1628,6 +1634,11 @@ func NewRouter(deps Dependencies) chi.Router {
 	} else {
 		downloadHandler = handlers.NewDownloadHandler(nil)
 	}
+	// Kill switch for in-flight download pours: a per-user stream revocation
+	// also hangs up a download mid-transfer (downloads stay exempt from the
+	// live-stream cap; this only closes the "revoked user keeps pulling a
+	// multi-GB file on a pre-revocation connection" hole).
+	downloadHandler.SetRevocationStore(deps.RevocationStore)
 
 	var policyHandler *handlers.PolicyHandler
 	if deps.PolicySystem != nil && deps.DB != nil {
@@ -2480,8 +2491,8 @@ func NewRouter(deps Dependencies) chi.Router {
 						// HLS transcode delivery — no profile auth needed;
 						// session ID (UUID) serves as the access token, same
 						// pattern as /stream/{session_id}.
-						r.Get("/transcode/{session_id}/master.m3u8", playbackHandler.HandleGetTranscodeManifest)
-						r.Get("/transcode/{session_id}/segment/{name}", playbackHandler.HandleGetTranscodeSegment)
+						r.Get("/transcode/{session_id}/master.m3u8", guardRevocation(deps.RevocationStore, configJWTSecret(deps), playbackHandler.HandleGetTranscodeManifest))
+						r.Get("/transcode/{session_id}/segment/{name}", guardRevocation(deps.RevocationStore, configJWTSecret(deps), playbackHandler.HandleGetTranscodeSegment))
 
 						// Playback realtime control socket — needs auth but not profile.
 						r.Get("/sessions/{session_id}/control/ws", playbackHandler.HandleSessionWebSocket)
@@ -2523,10 +2534,10 @@ func NewRouter(deps Dependencies) chi.Router {
 
 				// Stream routes.
 				if streamHandler != nil {
-					r.Get("/stream/{session_id}", streamHandler.HandleStream)
-					r.Head("/stream/{session_id}", streamHandler.HandleStream)
-					r.Get("/stream/{session_id}/subtitles/{track}", streamHandler.HandleSubtitle)
-					r.Get("/stream/{session_id}/subtitles/{track}/fonts", streamHandler.HandleSubtitleFonts)
+					r.Get("/stream/{session_id}", guardRevocationCut(deps.RevocationStore, configJWTSecret(deps), streamHandler.HandleStream))
+					r.Head("/stream/{session_id}", guardRevocationCut(deps.RevocationStore, configJWTSecret(deps), streamHandler.HandleStream))
+					r.Get("/stream/{session_id}/subtitles/{track}", guardRevocation(deps.RevocationStore, configJWTSecret(deps), streamHandler.HandleSubtitle))
+					r.Get("/stream/{session_id}/subtitles/{track}/fonts", guardRevocation(deps.RevocationStore, configJWTSecret(deps), streamHandler.HandleSubtitleFonts))
 				}
 
 				// Download routes.
@@ -2928,6 +2939,7 @@ func NewRouter(deps Dependencies) chi.Router {
 									jwtSecret = deps.Config.Auth.JWTSecret
 								}
 								nodeHandler := handlers.NewNodeHandler(deps.NodeRepo, deps.ProxyPool, deps.TranscodePool, deps.NodeRepo, deps.EventBus, deps.RedisClient, jwtSecret)
+								nodeHandler.SetLocalSessionSource(deps.SessionMgr, deps.NodeID)
 								r.Route("/nodes", func(r chi.Router) {
 									r.Get("/", nodeHandler.HandleListNodes)
 									r.Post("/", nodeHandler.HandleCreateNode)
@@ -3484,3 +3496,70 @@ func metadataAIConfigFromServer(cfg *config.Config) metadatatranslation.Config {
 		OnView:     cfg.MetadataAI.OnView,
 	}
 }
+
+// configJWTSecret returns the stream-token signing secret, or "" if unset.
+func configJWTSecret(deps Dependencies) string {
+	if deps.Config != nil {
+		return deps.Config.Auth.JWTSecret
+	}
+	return ""
+}
+
+// guardRevocation wraps a native (integrated-mode) stream serve handler so the
+// server refuses a revoked session/user before serving. Session id comes from
+// the {session_id} URL param; user id (best-effort) from the ?st= stream token.
+// This is the integrated-mode analogue of the edge nodes' IsRevoked check —
+// without it a kill (admin terminate, over-cap, account revocation) would have
+// no teeth on a single-node deployment. It refuses new/next requests (stopping
+// HLS on its next segment and refusing reconnects); cutting an in-flight native
+// direct-play pour is a follow-up.
+// streamRequestIdentity extracts the best-effort owner user id and the
+// credential-issue time for a native serve request. With a valid ?st= stream
+// token those are the token's uid and iat — the keys the user-kill cutoff in
+// streamrevoke compares (a token minted before a user revocation is killed; one
+// minted after re-auth plays). Without a token (legacy/bare URL or signing
+// disabled) these routes still run after API auth, so the authenticated user id
+// plus the request entry time apply: an entry request postdates any existing
+// user kill (fresh auth ⇒ allowed) while an in-flight pour predates a future
+// one (cut by WatchAndCut).
+func streamRequestIdentity(r *http.Request, secret string) (int, time.Time) {
+	if tok := r.URL.Query().Get("st"); tok != "" && secret != "" {
+		if claims, err := streamtoken.Verify(tok, secret); err == nil {
+			return claims.UserID, claims.IssuedTime()
+		}
+	}
+	return apimw.GetUserID(r.Context()), time.Now()
+}
+
+func guardRevocation(store *streamrevoke.Store, secret string, next http.HandlerFunc) http.HandlerFunc {
+	if store == nil {
+		return next
+	}
+	return func(w http.ResponseWriter, r *http.Request) {
+		userID, startedAt := streamRequestIdentity(r, secret)
+		if store.Refuse(w, chi.URLParam(r, "session_id"), userID, startedAt) {
+			return
+		}
+		next(w, r)
+	}
+}
+
+// guardRevocationCut is guardRevocation plus an in-flight connection cut, for the
+// long single-GET pours (native direct-play/remux on /stream). Transcode segment
+// routes use plain guardRevocation — per-segment refusal already stops HLS within
+// one segment, so a per-segment watcher goroutine would be wasteful.
+func guardRevocationCut(store *streamrevoke.Store, secret string, next http.HandlerFunc) http.HandlerFunc {
+	if store == nil {
+		return next
+	}
+	return func(w http.ResponseWriter, r *http.Request) {
+		sessionID := chi.URLParam(r, "session_id")
+		userID, startedAt := streamRequestIdentity(r, secret)
+		if store.Refuse(w, sessionID, userID, startedAt) {
+			return
+		}
+		stop := store.WatchAndCut(w, sessionID, userID, startedAt)
+		defer stop()
+		next(w, r)
+	}
+}