Publish ITWorx Pulse source

2026-09-03 02:09:19 +02:00
commit bd774932d5
614 changed files with 77116 additions and 0 deletions
@@ -0,0 +1,15 @@
+package live
+
+import "testing"
+
+func TestSampleOutputQueueDropsWhenFull(t *testing.T) {
+	session := &session{outbound: make(chan []byte, 1)}
+	session.outbound <- []byte("queued")
+	message := SamplesMessage{SchemaVersion: 1, Type: "samples", SubscriptionID: "sub", Sequence: 1, Samples: []Sample{}}
+	if err := session.send(message); err != nil {
+		t.Fatalf("sample drop should preserve session health: %v", err)
+	}
+	if len(session.outbound) != 1 {
+		t.Fatalf("queue length=%d, want bounded length 1", len(session.outbound))
+	}
+}
@@ -0,0 +1,511 @@
+package live
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+	"unicode/utf8"
+
+	"github.com/coder/websocket"
+	"github.com/itworx/pulse/internal/auth"
+	"github.com/itworx/pulse/internal/correlation"
+	"github.com/itworx/pulse/internal/queryplan"
+)
+
+const (
+	schemaVersion           = 1
+	defaultMaxMessage       = 64 << 10
+	defaultMaxSubs          = 50
+	defaultRateWindow       = time.Minute
+	defaultRateMessages     = 120
+	defaultHeartbeat        = 15 * time.Second
+	defaultIdleTimeout      = 45 * time.Second
+	defaultWriteTimeout     = 5 * time.Second
+	maxSubscriptionID       = 128
+	maxDetailLength         = 500
+	defaultOutboundMessages = 32
+)
+
+var (
+	ErrMessageTooLarge = errors.New("live message exceeds size limit")
+	ErrRateLimited     = errors.New("live message rate limit exceeded")
+)
+
+type Incoming struct {
+	SchemaVersion   int             `json:"schemaVersion"`
+	Type            string          `json:"type"`
+	SubscriptionID  string          `json:"subscriptionId,omitempty"`
+	Query           json.RawMessage `json:"query,omitempty"`
+	IntervalSeconds int             `json:"intervalSeconds,omitempty"`
+	Nonce           string          `json:"nonce,omitempty"`
+}
+
+type Sample struct {
+	Series    string            `json:"series"`
+	Timestamp time.Time         `json:"timestamp"`
+	Value     *float64          `json:"value"`
+	Freshness string            `json:"freshness"`
+	Labels    map[string]string `json:"labels,omitempty"`
+}
+
+type SamplesMessage struct {
+	SchemaVersion  int       `json:"schemaVersion"`
+	Type           string    `json:"type"`
+	SubscriptionID string    `json:"subscriptionId"`
+	Sequence       uint64    `json:"sequence"`
+	ServerTime     time.Time `json:"serverTime"`
+	Samples        []Sample  `json:"samples"`
+	Coalesced      bool      `json:"coalesced,omitempty"`
+}
+
+type StatusMessage struct {
+	SchemaVersion  int    `json:"schemaVersion"`
+	Type           string `json:"type"`
+	SubscriptionID string `json:"subscriptionId"`
+	State          string `json:"state"`
+	Detail         string `json:"detail,omitempty"`
+}
+
+type ErrorMessage struct {
+	SchemaVersion  int    `json:"schemaVersion"`
+	Type           string `json:"type"`
+	SubscriptionID string `json:"subscriptionId,omitempty"`
+	Code           string `json:"code"`
+	Message        string `json:"message"`
+	CorrelationID  string `json:"correlationId"`
+}
+
+type PongMessage struct {
+	SchemaVersion int    `json:"schemaVersion"`
+	Type          string `json:"type"`
+	Nonce         string `json:"nonce"`
+}
+
+type Sampler interface {
+	Sample(context.Context, queryplan.Request) ([]Sample, error)
+}
+
+type Handler struct {
+	Planner           *queryplan.Planner
+	Sampler           Sampler
+	AllowedOrigins    []string
+	MaxSubscriptions  int
+	MaxMessages       int
+	RateWindow        time.Duration
+	HeartbeatInterval time.Duration
+	IdleTimeout       time.Duration
+	WriteTimeout      time.Duration
+	Now               func() time.Time
+	Registry          *Registry
+}
+
+type subscription struct {
+	id       string
+	request  queryplan.Request
+	interval time.Duration
+	nextAt   time.Time
+	sequence uint64
+	lease    *Lease
+}
+
+type session struct {
+	conn          *websocket.Conn
+	principal     auth.Principal
+	planner       *queryplan.Planner
+	sampler       Sampler
+	maxSubs       int
+	maxMessages   int
+	rateWindow    time.Duration
+	heartbeat     time.Duration
+	idleTimeout   time.Duration
+	writeTimeout  time.Duration
+	registry      *Registry
+	now           func() time.Time
+	correlationID string
+	mu            sync.Mutex
+	subscriptions map[string]*subscription
+	rateStarted   time.Time
+	rateCount     int
+	outbound      chan []byte
+}
+
+func (h Handler) ServeHTTP(response http.ResponseWriter, request *http.Request) {
+	principal, ok := auth.PrincipalFromContext(request.Context())
+	if !ok {
+		http.Error(response, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+	if !auth.Allows(principal.Role, auth.PermissionView) {
+		http.Error(response, "forbidden", http.StatusForbidden)
+		return
+	}
+	conn, err := websocket.Accept(response, request, &websocket.AcceptOptions{OriginPatterns: h.AllowedOrigins})
+	if err != nil {
+		return
+	}
+	conn.SetReadLimit(int64(defaultMaxMessage))
+	registry := h.Registry
+	if registry == nil {
+		registry = NewRegistry(h.Sampler, RegistryOptions{})
+	}
+	// Keep the upgraded transport attached to the authenticated request. The
+	// session middleware cancels this context on logout and at the finite
+	// absolute session deadline; server/request cancellation must also release
+	// subscriptions and the socket.
+	ctx, cancel := context.WithCancel(request.Context())
+	correlationID := correlation.FromContext(request.Context())
+	if correlationID == "" {
+		correlationID = correlation.New()
+	}
+	s := &session{
+		conn: conn, principal: principal, planner: h.Planner, sampler: h.Sampler,
+		maxSubs:      positiveOr(h.MaxSubscriptions, defaultMaxSubs),
+		maxMessages:  positiveOr(h.MaxMessages, defaultRateMessages),
+		rateWindow:   durationOr(h.RateWindow, defaultRateWindow),
+		heartbeat:    durationOr(h.HeartbeatInterval, defaultHeartbeat),
+		idleTimeout:  durationOr(h.IdleTimeout, defaultIdleTimeout),
+		writeTimeout: durationOr(h.WriteTimeout, defaultWriteTimeout),
+		registry:     registry,
+		now:          h.Now, correlationID: correlationID, subscriptions: make(map[string]*subscription), outbound: make(chan []byte, defaultOutboundMessages),
+	}
+	if s.now == nil {
+		s.now = func() time.Time { return time.Now().UTC() }
+	}
+	defer func() {
+		cancel()
+		s.releaseAll()
+		conn.CloseNow()
+	}()
+	go s.writeLoop(ctx, cancel)
+	go s.heartbeatLoop(ctx, cancel)
+	go s.sampleLoop(ctx, cancel)
+	s.readLoop(ctx)
+}
+
+func positiveOr(value, fallback int) int {
+	if value > 0 {
+		return value
+	}
+	return fallback
+}
+
+func durationOr(value, fallback time.Duration) time.Duration {
+	if value > 0 {
+		return value
+	}
+	return fallback
+}
+
+func (s *session) readLoop(ctx context.Context) {
+	for {
+		readCtx, cancel := context.WithTimeout(ctx, s.idleTimeout)
+		_, payload, err := s.conn.Read(readCtx)
+		cancel()
+		if err != nil {
+			return
+		}
+		if err := s.acceptMessage(ctx, payload); err != nil {
+			code, detail := "LIVE_MESSAGE_INVALID", "Live bericht is ongeldig."
+			if errors.Is(err, ErrMessageTooLarge) {
+				code, detail = "LIVE_MESSAGE_LIMIT", "Live bericht overschrijdt de maximale berichtgrootte."
+			} else if errors.Is(err, ErrRateLimited) {
+				code, detail = "LIVE_RATE_LIMIT", "Te veel live berichten; verbind opnieuw na een korte pauze."
+			}
+			_ = s.send(ErrorMessage{SchemaVersion: schemaVersion, Type: "error", Code: code, Message: detail, CorrelationID: s.correlationID})
+			return
+		}
+	}
+}
+
+func (s *session) acceptMessage(ctx context.Context, payload []byte) error {
+	if len(payload) > defaultMaxMessage {
+		return ErrMessageTooLarge
+	}
+	s.mu.Lock()
+	now := s.now()
+	if s.rateStarted.IsZero() || now.Sub(s.rateStarted) >= s.rateWindow {
+		s.rateStarted = now
+		s.rateCount = 0
+	}
+	s.rateCount++
+	limited := s.rateCount > s.maxMessages
+	s.mu.Unlock()
+	if limited {
+		return ErrRateLimited
+	}
+	var message Incoming
+	decoder := json.NewDecoder(bytes.NewReader(payload))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&message); err != nil {
+		return err
+	}
+	if err := decoder.Decode(&struct{}{}); err != io.EOF {
+		return errors.New("live message has trailing data")
+	}
+	if message.SchemaVersion != schemaVersion || !validTypeFields(payload, message.Type) {
+		return errors.New("live message schema is invalid")
+	}
+	switch message.Type {
+	case "subscribe":
+		return s.subscribe(ctx, message)
+	case "unsubscribe":
+		return s.unsubscribe(message.SubscriptionID)
+	case "ping":
+		if !validNonce(message.Nonce) {
+			return errors.New("ping nonce is invalid")
+		}
+		return s.send(PongMessage{SchemaVersion: schemaVersion, Type: "pong", Nonce: message.Nonce})
+	case "pong":
+		if !validNonce(message.Nonce) {
+			return errors.New("pong nonce is invalid")
+		}
+		return nil
+	default:
+		return errors.New("live message type is unsupported")
+	}
+}
+
+func validTypeFields(payload []byte, typ string) bool {
+	var fields map[string]json.RawMessage
+	if json.Unmarshal(payload, &fields) != nil {
+		return false
+	}
+	allowed := map[string]map[string]struct{}{
+		"subscribe":   {"schemaVersion": {}, "type": {}, "subscriptionId": {}, "query": {}, "intervalSeconds": {}},
+		"unsubscribe": {"schemaVersion": {}, "type": {}, "subscriptionId": {}},
+		"ping":        {"schemaVersion": {}, "type": {}, "nonce": {}},
+		"pong":        {"schemaVersion": {}, "type": {}, "nonce": {}},
+	}
+	known, ok := allowed[typ]
+	if !ok {
+		return false
+	}
+	for key := range fields {
+		if _, ok := known[key]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+func validSubscriptionID(id string) bool {
+	return id != "" && utf8.RuneCountInString(id) <= maxSubscriptionID
+}
+
+func validNonce(nonce string) bool {
+	return utf8.RuneCountInString(nonce) <= maxSubscriptionID
+}
+
+func (s *session) subscribe(ctx context.Context, message Incoming) error {
+	if !validSubscriptionID(message.SubscriptionID) || len(message.Query) == 0 || message.IntervalSeconds < 1 || message.IntervalSeconds > 300 {
+		return s.sendError(message.SubscriptionID, "LIVE_SUBSCRIPTION_INVALID", "Abonnement, query of interval is ongeldig.")
+	}
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(message.Query, &fields); err != nil || len(fields) > 20 {
+		return s.sendError(message.SubscriptionID, "LIVE_QUERY_INVALID", "Live query moet een begrensd JSON-object zijn.")
+	}
+	var request queryplan.Request
+	decoder := json.NewDecoder(bytes.NewReader(message.Query))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&request); err != nil {
+		return s.sendError(message.SubscriptionID, "LIVE_QUERY_INVALID", "Live query bevat onbekende of ongeldige velden.")
+	}
+	if s.planner == nil {
+		return s.sendError(message.SubscriptionID, "LIVE_QUERY_UNAVAILABLE", "Live queryvalidatie is tijdelijk niet beschikbaar.")
+	}
+	plan, err := s.planner.Plan(auth.WithPrincipal(ctx, s.principal), request)
+	if err != nil {
+		return s.sendError(message.SubscriptionID, "LIVE_SUBSCRIPTION_DENIED", queryErrorDetail(err))
+	}
+	now := s.now()
+	interval := time.Duration(message.IntervalSeconds) * time.Second
+	s.mu.Lock()
+	duplicate := false
+	sessionLimit := len(s.subscriptions) >= s.maxSubs
+	if _, exists := s.subscriptions[message.SubscriptionID]; exists {
+		duplicate = true
+	}
+	s.mu.Unlock()
+	if duplicate {
+		return s.sendError(message.SubscriptionID, "LIVE_SUBSCRIPTION_CONFLICT", "Subscription-ID bestaat al.")
+	}
+	if sessionLimit {
+		return s.sendError(message.SubscriptionID, "LIVE_SUBSCRIPTION_LIMIT", "Maximum aantal live abonnementen bereikt.")
+	}
+	lease, err := s.registry.Acquire(plan.Request, interval)
+	if err != nil {
+		if errors.Is(err, ErrRegistryLimit) {
+			return s.sendError(message.SubscriptionID, "LIVE_REGISTRY_LIMIT", "Maximum aantal gedeelde live queries bereikt.")
+		}
+		return s.sendError(message.SubscriptionID, "LIVE_QUERY_INVALID", "Live query kon niet worden genormaliseerd.")
+	}
+	s.mu.Lock()
+	if _, exists := s.subscriptions[message.SubscriptionID]; exists || len(s.subscriptions) >= s.maxSubs {
+		s.mu.Unlock()
+		lease.Release()
+		return s.sendError(message.SubscriptionID, "LIVE_SUBSCRIPTION_CONFLICT", "Subscription-ID bestaat al.")
+	}
+	s.subscriptions[message.SubscriptionID] = &subscription{id: message.SubscriptionID, request: plan.Request, interval: interval, nextAt: now.Add(interval), lease: lease}
+	s.mu.Unlock()
+	return s.send(StatusMessage{SchemaVersion: schemaVersion, Type: "status", SubscriptionID: message.SubscriptionID, State: "subscribed"})
+}
+
+func queryErrorDetail(err error) string {
+	var plannerError queryplan.Error
+	if errors.As(err, &plannerError) {
+		return plannerError.Detail
+	}
+	return "Live query kon niet worden geautoriseerd."
+}
+
+func (s *session) unsubscribe(id string) error {
+	if !validSubscriptionID(id) {
+		return s.sendError(id, "LIVE_SUBSCRIPTION_INVALID", "Subscription-ID is ongeldig.")
+	}
+	s.mu.Lock()
+	sub, exists := s.subscriptions[id]
+	delete(s.subscriptions, id)
+	s.mu.Unlock()
+	if exists && sub.lease != nil {
+		sub.lease.Release()
+	}
+	return s.send(StatusMessage{SchemaVersion: schemaVersion, Type: "status", SubscriptionID: id, State: "unsubscribed"})
+}
+
+func (s *session) releaseAll() {
+	s.mu.Lock()
+	subscriptions := make([]*subscription, 0, len(s.subscriptions))
+	for _, sub := range s.subscriptions {
+		subscriptions = append(subscriptions, sub)
+	}
+	s.subscriptions = make(map[string]*subscription)
+	s.mu.Unlock()
+	for _, sub := range subscriptions {
+		if sub.lease != nil {
+			sub.lease.Release()
+		}
+	}
+}
+
+func (s *session) sendError(subscriptionID, code, message string) error {
+	if len(message) > maxDetailLength {
+		message = message[:maxDetailLength]
+	}
+	return s.send(ErrorMessage{SchemaVersion: schemaVersion, Type: "error", SubscriptionID: subscriptionID, Code: code, Message: message, CorrelationID: s.correlationID})
+}
+
+func (s *session) send(message any) error {
+	payload, err := json.Marshal(message)
+	if err != nil {
+		return err
+	}
+	if len(payload) > defaultMaxMessage {
+		return ErrMessageTooLarge
+	}
+	sample := false
+	switch message.(type) {
+	case SamplesMessage, *SamplesMessage:
+		sample = true
+	}
+	if !sample {
+		ctx, cancel := context.WithTimeout(context.Background(), s.writeTimeout)
+		defer cancel()
+		return s.conn.Write(ctx, websocket.MessageText, payload)
+	}
+	if s.outbound == nil {
+		return ErrOutboundBackpressure
+	}
+	select {
+	case s.outbound <- payload:
+		return nil
+	default:
+		return nil
+	}
+}
+
+func (s *session) writeLoop(ctx context.Context, cancel context.CancelFunc) {
+	for {
+		select {
+		case payload := <-s.outbound:
+			writeCtx, writeCancel := context.WithTimeout(ctx, s.writeTimeout)
+			err := s.conn.Write(writeCtx, websocket.MessageText, payload)
+			writeCancel()
+			if err != nil {
+				cancel()
+				return
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (s *session) heartbeatLoop(ctx context.Context, cancel context.CancelFunc) {
+	ticker := time.NewTicker(s.heartbeat)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			pingCtx, pingCancel := context.WithTimeout(ctx, s.writeTimeout)
+			err := s.conn.Ping(pingCtx)
+			pingCancel()
+			if err != nil {
+				cancel()
+				return
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (s *session) sampleLoop(ctx context.Context, cancel context.CancelFunc) {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			now := s.now()
+			s.mu.Lock()
+			due := make([]*subscription, 0, len(s.subscriptions))
+			for _, sub := range s.subscriptions {
+				if !now.Before(sub.nextAt) {
+					sub.nextAt = now.Add(sub.interval)
+					due = append(due, sub)
+				}
+			}
+			s.mu.Unlock()
+			for _, sub := range due {
+				samples, err := s.registry.Sample(ctx, sub.request)
+				if err != nil {
+					_ = s.sendError(sub.id, "LIVE_SAMPLE_UNAVAILABLE", "Live sample is tijdelijk niet beschikbaar.")
+					continue
+				}
+				s.mu.Lock()
+				current, exists := s.subscriptions[sub.id]
+				if !exists {
+					s.mu.Unlock()
+					continue
+				}
+				current.sequence++
+				sequence := current.sequence
+				s.mu.Unlock()
+				if len(samples) > 10000 {
+					_ = s.sendError(sub.id, "LIVE_SAMPLE_LIMIT", "Live sample bevat te veel punten.")
+					continue
+				}
+				if err := s.send(SamplesMessage{SchemaVersion: schemaVersion, Type: "samples", SubscriptionID: sub.id, Sequence: sequence, ServerTime: now.UTC(), Samples: samples}); err != nil {
+					cancel()
+					return
+				}
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
@@ -0,0 +1,222 @@
+package live_test
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/wsjson"
+	"github.com/itworx/pulse/internal/auth"
+	"github.com/itworx/pulse/internal/live"
+	"github.com/itworx/pulse/internal/metriccatalog"
+	"github.com/itworx/pulse/internal/queryplan"
+)
+
+type sampler struct{}
+
+func (sampler) Sample(_ context.Context, request queryplan.Request) ([]live.Sample, error) {
+	value := float64(request.MaxPoints)
+	return []live.Sample{{Series: "test", Timestamp: time.Now().UTC(), Value: &value, Freshness: "fresh"}}, nil
+}
+
+func plannerForTest(t *testing.T) *queryplan.Planner {
+	t.Helper()
+	registry, err := metriccatalog.DefaultRegistry()
+	if err != nil {
+		t.Fatal(err)
+	}
+	planner := queryplan.NewPlanner(registry, queryplan.Limits{})
+	return &planner
+}
+
+func validQuery() map[string]any {
+	now := time.Now().UTC().Truncate(time.Second)
+	return map[string]any{
+		"metric":      "container.cpu.utilization",
+		"scope":       map[string]string{"containerId": "media_server"},
+		"range":       map[string]any{"from": now.Add(-time.Minute).Format(time.RFC3339), "to": now.Format(time.RFC3339), "stepSeconds": 15},
+		"aggregation": "avg",
+		"maxSeries":   1,
+		"maxPoints":   60,
+	}
+}
+
+func authorizedServer(t *testing.T, handler live.Handler) *httptest.Server {
+	t.Helper()
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		r = r.WithContext(auth.WithPrincipal(r.Context(), auth.Principal{Subject: "viewer", Role: auth.RoleViewer}))
+		handler.ServeHTTP(w, r)
+	}))
+}
+
+func TestHandlerRejectsUnauthenticatedBeforeUpgrade(t *testing.T) {
+	server := httptest.NewServer(live.Handler{})
+	defer server.Close()
+	_, response, err := websocket.Dial(context.Background(), "ws"+server.URL[4:]+"/api/v1/live", nil)
+	if err == nil {
+		t.Fatal("expected unauthenticated upgrade rejection")
+	}
+	if response == nil || response.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("response=%v err=%v", response, err)
+	}
+}
+
+func TestHandlerRejectsCrossOriginUpgrade(t *testing.T) {
+	server := authorizedServer(t, live.Handler{Planner: plannerForTest(t)})
+	defer server.Close()
+	options := &websocket.DialOptions{HTTPHeader: http.Header{"Origin": []string{"https://evil.example"}}}
+	_, response, err := websocket.Dial(context.Background(), "ws"+server.URL[4:]+"/api/v1/live", options)
+	if err == nil {
+		t.Fatal("expected cross-origin rejection")
+	}
+	if response == nil || response.StatusCode != http.StatusForbidden {
+		t.Fatalf("response=%v err=%v", response, err)
+	}
+}
+
+func TestHandlerClosesWhenAuthenticatedRequestIsCancelled(t *testing.T) {
+	cancelled := make(chan context.CancelFunc, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx, cancel := context.WithCancel(auth.WithPrincipal(r.Context(), auth.Principal{Subject: "viewer", Role: auth.RoleViewer}))
+		cancelled <- cancel
+		live.Handler{}.ServeHTTP(w, r.WithContext(ctx))
+	}))
+	defer server.Close()
+	conn, _, err := websocket.Dial(context.Background(), "ws"+server.URL[4:]+"/api/v1/live", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.CloseNow()
+	(<-cancelled)()
+	readCtx, cancelRead := context.WithTimeout(context.Background(), time.Second)
+	defer cancelRead()
+	if _, _, err := conn.Read(readCtx); err == nil {
+		t.Fatal("live connection survived authenticated request cancellation")
+	}
+}
+
+func TestHandlerAuthorizesSubscriptionAndSequencesSamples(t *testing.T) {
+	server := authorizedServer(t, live.Handler{Planner: plannerForTest(t), Sampler: sampler{}})
+	defer server.Close()
+	conn, _, err := websocket.Dial(context.Background(), "ws"+server.URL[4:]+"/api/v1/live", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close(websocket.StatusNormalClosure, "")
+	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
+	defer cancel()
+	if err := wsjson.Write(ctx, conn, map[string]any{"schemaVersion": 1, "type": "subscribe", "subscriptionId": "sub-1", "query": validQuery(), "intervalSeconds": 1}); err != nil {
+		t.Fatal(err)
+	}
+	var status live.StatusMessage
+	if err := wsjson.Read(ctx, conn, &status); err != nil {
+		t.Fatal(err)
+	}
+	if status.State != "subscribed" {
+		t.Fatalf("status=%+v", status)
+	}
+	var first, second live.SamplesMessage
+	if err := wsjson.Read(ctx, conn, &first); err != nil {
+		t.Fatal(err)
+	}
+	if err := wsjson.Read(ctx, conn, &second); err != nil {
+		t.Fatal(err)
+	}
+	if first.Sequence != 1 || second.Sequence != 2 || second.Sequence <= first.Sequence {
+		t.Fatalf("sequences=%d,%d", first.Sequence, second.Sequence)
+	}
+	if len(first.Samples) != 1 || first.Samples[0].Freshness != "fresh" {
+		t.Fatalf("samples=%+v", first.Samples)
+	}
+}
+
+func TestHandlerReportsMissingSamplerInsteadOfEmptySamples(t *testing.T) {
+	server := authorizedServer(t, live.Handler{Planner: plannerForTest(t)})
+	defer server.Close()
+	conn, _, err := websocket.Dial(context.Background(), "ws"+server.URL[4:]+"/api/v1/live", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = conn.Close(websocket.StatusNormalClosure, "") }()
+	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
+	defer cancel()
+	if err := wsjson.Write(ctx, conn, map[string]any{"schemaVersion": 1, "type": "subscribe", "subscriptionId": "sub-1", "query": validQuery(), "intervalSeconds": 1}); err != nil {
+		t.Fatal(err)
+	}
+	var status live.StatusMessage
+	if err := wsjson.Read(ctx, conn, &status); err != nil {
+		t.Fatal(err)
+	}
+	if status.State != "subscribed" {
+		t.Fatalf("status=%+v", status)
+	}
+	var failure live.ErrorMessage
+	if err := wsjson.Read(ctx, conn, &failure); err != nil {
+		t.Fatal(err)
+	}
+	if failure.Type != "error" || failure.Code != "LIVE_SAMPLE_UNAVAILABLE" || failure.SubscriptionID != "sub-1" {
+		t.Fatalf("failure=%+v", failure)
+	}
+}
+
+func TestHandlerDeniesInvalidSubscriptionAndLimitsMessages(t *testing.T) {
+	server := authorizedServer(t, live.Handler{Planner: plannerForTest(t), MaxMessages: 2, RateWindow: time.Minute})
+	defer server.Close()
+	conn, _, err := websocket.Dial(context.Background(), "ws"+server.URL[4:]+"/api/v1/live", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close(websocket.StatusNormalClosure, "")
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	if err := wsjson.Write(ctx, conn, map[string]any{"schemaVersion": 1, "type": "subscribe", "subscriptionId": "bad", "query": map[string]any{"metric": "unknown"}, "intervalSeconds": 1}); err != nil {
+		t.Fatal(err)
+	}
+	var problem live.ErrorMessage
+	if err := wsjson.Read(ctx, conn, &problem); err != nil {
+		t.Fatal(err)
+	}
+	if problem.Code != "LIVE_SUBSCRIPTION_DENIED" {
+		t.Fatalf("problem=%+v", problem)
+	}
+	if err := wsjson.Write(ctx, conn, map[string]any{"schemaVersion": 1, "type": "ping", "nonce": "one"}); err != nil {
+		t.Fatal(err)
+	}
+	var pong live.PongMessage
+	if err := wsjson.Read(ctx, conn, &pong); err != nil {
+		t.Fatal(err)
+	}
+	if err := wsjson.Write(ctx, conn, map[string]any{"schemaVersion": 1, "type": "ping", "nonce": "two"}); err != nil {
+		t.Fatal(err)
+	}
+	var limited live.ErrorMessage
+	if err := wsjson.Read(ctx, conn, &limited); err != nil {
+		t.Fatal(err)
+	}
+	if limited.Code != "LIVE_RATE_LIMIT" {
+		t.Fatalf("problem=%+v", limited)
+	}
+}
+
+func TestHandlerClosesOversizedMessage(t *testing.T) {
+	server := authorizedServer(t, live.Handler{Planner: plannerForTest(t)})
+	defer server.Close()
+	conn, _, err := websocket.Dial(context.Background(), "ws"+server.URL[4:]+"/api/v1/live", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close(websocket.StatusNormalClosure, "")
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	if err := wsjson.Write(ctx, conn, map[string]any{"schemaVersion": 1, "type": "ping", "nonce": strings.Repeat("x", 70000)}); err != nil {
+		t.Fatal(err)
+	}
+	var response map[string]any
+	if err := wsjson.Read(ctx, conn, &response); err == nil {
+		t.Fatal("expected oversized message connection to close")
+	}
+}
@@ -0,0 +1,243 @@
+package live
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/itworx/pulse/internal/queryplan"
+)
+
+const (
+	defaultRegistryEntries     = 256
+	defaultRegistryConcurrency = 16
+)
+
+var (
+	ErrRegistryLimit        = errors.New("live subscription registry limit reached")
+	ErrLeaseReleased        = errors.New("live subscription lease is released")
+	ErrOutboundBackpressure = errors.New("live outbound queue is full")
+	ErrSamplerUnavailable   = errors.New("live sampler is not configured")
+)
+
+type RegistryOptions struct {
+	MaxEntries    int
+	MaxConcurrent int
+}
+
+type Registry struct {
+	sampler    Sampler
+	now        func() time.Time
+	maxEntries int
+	sem        chan struct{}
+	mu         sync.Mutex
+	entries    map[string]*registryEntry
+}
+
+type registryEntry struct {
+	key         string
+	request     queryplan.Request
+	references  int
+	minInterval time.Duration
+	lastAt      time.Time
+	lastSamples []Sample
+	inFlight    *sampleCall
+}
+
+type sampleCall struct {
+	done    chan struct{}
+	samples []Sample
+	err     error
+}
+
+type Lease struct {
+	registry *Registry
+	key      string
+	once     sync.Once
+}
+
+// NewRegistry builds the shared live subscription registry. A nil sampler keeps
+// subscription lifecycle traffic working but is not a usable data source: every
+// Sample then fails with ErrSamplerUnavailable instead of silently reporting an
+// empty successful result.
+func NewRegistry(sampler Sampler, options RegistryOptions) *Registry {
+	maxEntries := options.MaxEntries
+	if maxEntries <= 0 {
+		maxEntries = defaultRegistryEntries
+	}
+	maxConcurrent := options.MaxConcurrent
+	if maxConcurrent <= 0 {
+		maxConcurrent = defaultRegistryConcurrency
+	}
+	return &Registry{
+		sampler:    sampler,
+		now:        func() time.Time { return time.Now().UTC() },
+		maxEntries: maxEntries,
+		sem:        make(chan struct{}, maxConcurrent),
+		entries:    make(map[string]*registryEntry),
+	}
+}
+
+func (r *Registry) Acquire(request queryplan.Request, interval time.Duration) (*Lease, error) {
+	if r == nil || interval <= 0 {
+		return nil, ErrLeaseReleased
+	}
+	key, err := normalizedKey(request)
+	if err != nil {
+		return nil, err
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	entry, exists := r.entries[key]
+	if !exists {
+		if len(r.entries) >= r.maxEntries {
+			return nil, ErrRegistryLimit
+		}
+		entry = &registryEntry{key: key, request: request, minInterval: interval}
+		r.entries[key] = entry
+	} else if interval < entry.minInterval {
+		entry.minInterval = interval
+	}
+	entry.references++
+	return &Lease{registry: r, key: key}, nil
+}
+
+func (l *Lease) Release() {
+	if l == nil || l.registry == nil {
+		return
+	}
+	l.once.Do(func() {
+		l.registry.release(l.key)
+	})
+}
+
+func (r *Registry) release(key string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	entry, ok := r.entries[key]
+	if !ok {
+		return
+	}
+	if entry.references > 0 {
+		entry.references--
+	}
+	if entry.references == 0 && entry.inFlight == nil {
+		delete(r.entries, key)
+	}
+}
+
+func (r *Registry) Sample(ctx context.Context, request queryplan.Request) ([]Sample, error) {
+	if r == nil {
+		return nil, ErrLeaseReleased
+	}
+	if r.sampler == nil {
+		return nil, ErrSamplerUnavailable
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	key, err := normalizedKey(request)
+	if err != nil {
+		return nil, err
+	}
+	r.mu.Lock()
+	entry, ok := r.entries[key]
+	if !ok || entry.references == 0 {
+		r.mu.Unlock()
+		return nil, ErrLeaseReleased
+	}
+	now := r.now()
+	if !entry.lastAt.IsZero() && now.Sub(entry.lastAt) < entry.minInterval {
+		samples := cloneSamples(entry.lastSamples)
+		r.mu.Unlock()
+		return samples, nil
+	}
+	if entry.inFlight != nil {
+		call := entry.inFlight
+		r.mu.Unlock()
+		select {
+		case <-call.done:
+			return cloneSamples(call.samples), call.err
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+	call := &sampleCall{done: make(chan struct{})}
+	entry.inFlight = call
+	requestCopy := entry.request
+	r.mu.Unlock()
+
+	select {
+	case r.sem <- struct{}{}:
+	case <-ctx.Done():
+		r.finish(key, entry, call, nil, ctx.Err())
+		return nil, ctx.Err()
+	}
+	samples, sampleErr := r.sampler.Sample(ctx, requestCopy)
+	<-r.sem
+	r.finish(key, entry, call, samples, sampleErr)
+	return cloneSamples(samples), sampleErr
+}
+
+func (r *Registry) finish(key string, entry *registryEntry, call *sampleCall, samples []Sample, err error) {
+	r.mu.Lock()
+	call.samples = cloneSamples(samples)
+	call.err = err
+	if err == nil {
+		entry.lastAt = r.now()
+		entry.lastSamples = cloneSamples(samples)
+	}
+	entry.inFlight = nil
+	if entry.references == 0 {
+		delete(r.entries, key)
+	}
+	close(call.done)
+	r.mu.Unlock()
+}
+
+func (r *Registry) Active() (entries, references int) {
+	if r == nil {
+		return 0, 0
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, entry := range r.entries {
+		entries++
+		references += entry.references
+	}
+	return entries, references
+}
+
+func normalizedKey(request queryplan.Request) (string, error) {
+	payload, err := json.Marshal(request)
+	if err != nil {
+		return "", err
+	}
+	digest := sha256.Sum256(payload)
+	return hex.EncodeToString(digest[:]), nil
+}
+
+func cloneSamples(samples []Sample) []Sample {
+	if samples == nil {
+		return nil
+	}
+	cloned := make([]Sample, len(samples))
+	for index, sample := range samples {
+		cloned[index] = sample
+		if sample.Value != nil {
+			value := *sample.Value
+			cloned[index].Value = &value
+		}
+		if sample.Labels != nil {
+			cloned[index].Labels = make(map[string]string, len(sample.Labels))
+			for key, value := range sample.Labels {
+				cloned[index].Labels[key] = value
+			}
+		}
+	}
+	return cloned
+}
@@ -0,0 +1,185 @@
+package live_test
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/itworx/pulse/internal/live"
+	"github.com/itworx/pulse/internal/queryplan"
+)
+
+type blockingSampler struct {
+	mu      sync.Mutex
+	calls   int
+	started chan struct{}
+	release chan struct{}
+}
+
+func (s *blockingSampler) Sample(_ context.Context, _ queryplan.Request) ([]live.Sample, error) {
+	s.mu.Lock()
+	s.calls++
+	if s.calls == 1 {
+		close(s.started)
+	}
+	s.mu.Unlock()
+	<-s.release
+	value := 1.0
+	return []live.Sample{{Series: "shared", Timestamp: time.Now().UTC(), Value: &value, Freshness: "fresh"}}, nil
+}
+
+func (s *blockingSampler) Calls() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.calls
+}
+
+func sharedRequest() queryplan.Request {
+	now := time.Now().UTC().Truncate(time.Second)
+	return queryplan.Request{
+		Metric:      "container.cpu.utilization",
+		Scope:       map[string]string{"container": "media_server"},
+		Range:       queryplan.Range{From: now.Add(-time.Minute), To: now, StepSeconds: 15},
+		Aggregation: "avg",
+		MaxSeries:   1,
+		MaxPoints:   60,
+	}
+}
+
+func TestRegistryDeduplicatesOverlappingSamplesAndReleasesReferences(t *testing.T) {
+	source := &blockingSampler{started: make(chan struct{}), release: make(chan struct{})}
+	registry := live.NewRegistry(source, live.RegistryOptions{MaxEntries: 4, MaxConcurrent: 1})
+	request := sharedRequest()
+	firstLease, err := registry.Acquire(request, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondLease, err := registry.Acquire(request, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	entries, references := registry.Active()
+	if entries != 1 || references != 2 {
+		t.Fatalf("active=%d references=%d", entries, references)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	first := make(chan error, 1)
+	go func() {
+		_, err := registry.Sample(ctx, request)
+		first <- err
+	}()
+	<-source.started
+	second := make(chan error, 1)
+	go func() {
+		_, err := registry.Sample(ctx, request)
+		second <- err
+	}()
+	select {
+	case err := <-second:
+		t.Fatalf("second sample completed before shared source: %v", err)
+	case <-time.After(25 * time.Millisecond):
+	}
+	close(source.release)
+	if err := <-first; err != nil {
+		t.Fatal(err)
+	}
+	if err := <-second; err != nil {
+		t.Fatal(err)
+	}
+	if calls := source.Calls(); calls != 1 {
+		t.Fatalf("upstream calls=%d, want 1", calls)
+	}
+
+	firstLease.Release()
+	entries, references = registry.Active()
+	if entries != 1 || references != 1 {
+		t.Fatalf("after first release active=%d references=%d", entries, references)
+	}
+	secondLease.Release()
+	entries, references = registry.Active()
+	if entries != 0 || references != 0 {
+		t.Fatalf("after final release active=%d references=%d", entries, references)
+	}
+}
+
+func TestRegistryWithoutSamplerFailsVisiblyInsteadOfReturningEmptySamples(t *testing.T) {
+	registry := live.NewRegistry(nil, live.RegistryOptions{})
+	request := sharedRequest()
+	lease, err := registry.Acquire(request, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lease.Release()
+	samples, err := registry.Sample(context.Background(), request)
+	if !errors.Is(err, live.ErrSamplerUnavailable) {
+		t.Fatalf("nil sampler reported err=%v", err)
+	}
+	if samples != nil {
+		t.Fatalf("nil sampler returned samples=%+v", samples)
+	}
+}
+
+type trackingSampler struct {
+	mu        sync.Mutex
+	active    int
+	maxActive int
+	started   chan struct{}
+	release   chan struct{}
+}
+
+func (s *trackingSampler) Sample(_ context.Context, _ queryplan.Request) ([]live.Sample, error) {
+	s.mu.Lock()
+	s.active++
+	if s.active > s.maxActive {
+		s.maxActive = s.active
+	}
+	s.mu.Unlock()
+	s.started <- struct{}{}
+	<-s.release
+	s.mu.Lock()
+	s.active--
+	s.mu.Unlock()
+	return []live.Sample{}, nil
+}
+
+func (s *trackingSampler) MaxActive() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.maxActive
+}
+
+func TestRegistryBoundsEntriesAndUpstreamConcurrency(t *testing.T) {
+	source := &trackingSampler{started: make(chan struct{}, 2), release: make(chan struct{})}
+	registry := live.NewRegistry(source, live.RegistryOptions{MaxEntries: 1, MaxConcurrent: 1})
+	firstRequest := sharedRequest()
+	firstLease, err := registry.Acquire(firstRequest, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer firstLease.Release()
+	secondRequest := firstRequest
+	secondRequest.Metric = "host.cpu.utilization"
+	if _, err := registry.Acquire(secondRequest, time.Second); err == nil {
+		t.Fatal("expected registry entry limit")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	first := make(chan error, 1)
+	go func() {
+		_, err := registry.Sample(ctx, firstRequest)
+		first <- err
+	}()
+	<-source.started
+	close(source.release)
+	if err := <-first; err != nil {
+		t.Fatal(err)
+	}
+	if source.MaxActive() != 1 {
+		t.Fatalf("max upstream concurrency=%d", source.MaxActive())
+	}
+}