Edu/services/push-gateway/internal/sse/handler.go

// Package sse implements the Server-Sent Events endpoint for real-time
// notification delivery.
//
// M7 (ADR-040): The SSE endpoint replaces direct Kafka mounting with a
// Redis Pub/Sub backplane. Each SSE connection dynamically subscribes to
// the Redis channel `user:{userId}:notify` for the authenticated user.
// When the user disconnects, the subscription is released. This per-connection
// subscription model means each realtime-gateway instance only holds Redis
// subscriptions for users actively connected to it, enabling horizontal
// scaling behind a load balancer.
//
// Authentication:
//   - JWT RS256 validated via shared-go/jwks (caching JWKS from iam
//     /.well-known/jwks.json). DevMode accepts the literal "dev-token"
//     returning a synthetic dev-user subject.
//
// SSE format: `data: {json}\n\n` (per the HTML5 Server-Sent Events spec).
// Heartbeat comments (`: heartbeat\n\n`) are sent every 30s to keep the
// connection alive through proxies.
package sse

import (
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"github.com/edu-cloud/push-gateway/internal/config"
	"github.com/edu-cloud/push-gateway/internal/observability"
	"github.com/edu-cloud/push-gateway/internal/redisclient"
	"github.com/edu-cloud/shared-go/jwks"
	"github.com/gin-gonic/gin"
	"github.com/golang-jwt/jwt/v5"
)

// heartbeatInterval is how often a comment frame is sent to keep the SSE
// connection alive through proxies that idle-timeout connections.
const heartbeatInterval = 30 * time.Second

// errResponse is the ActionState-shaped error envelope returned for auth
// failures and Redis unavailability.
type errResponse struct {
	Success bool    `json:"success"`
	Error   *errBody `json:"error,omitempty"`
}

type errBody struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

var (
	errMissingToken  = errors.New("missing token")
	errInvalidToken  = errors.New("invalid token")
	errMissingUserID = errors.New("missing user id")
)

// Handler owns the SSE endpoint for notification delivery.
type Handler struct {
	cfg         *config.Config
	redis       *redisclient.Client
	jwksFetcher *jwks.Fetcher
	metrics     *observability.Metrics
}

// NewHandler constructs an SSE Handler. cfg supplies JWT auth configuration;
// redis may be nil in DevMode (SSE returns 503 when Redis is unavailable).
// jwksFetcher is initialized from cfg.JWKSURL when non-empty.
func NewHandler(cfg *config.Config, r *redisclient.Client, m *observability.Metrics) *Handler {
	var fetcher *jwks.Fetcher
	if cfg.JWKSURL != "" {
		fetcher = jwks.NewFetcher(cfg.JWKSURL)
	}
	return &Handler{
		cfg:         cfg,
		redis:       r,
		jwksFetcher: fetcher,
		metrics:     m,
	}
}

// HandleNotifications is the GET /sse/notifications endpoint. It upgrades the
// HTTP connection to a persistent SSE stream, subscribing to the user's
// Redis Pub/Sub channel and forwarding messages as SSE data frames.
//
// Lifecycle:
//  1. Authenticate via JWT (query ?token= or Authorization: Bearer header)
//  2. Set SSE headers (Content-Type: text/event-stream, no-cache, etc.)
//  3. SUBSCRIBE to Redis channel `user:{userId}:notify`
//  4. Forward each Redis message as `data: {json}\n\n`
//  5. Send heartbeat comment every 30s
//  6. On client disconnect or context cancel: UNSUBSCRIBE (deferred pubsub.Close)
func (h *Handler) HandleNotifications(c *gin.Context) {
	userID, err := h.authenticate(c)
	if err != nil {
		c.AbortWithStatusJSON(http.StatusUnauthorized, errResponse{
			Success: false,
			Error:   &errBody{Code: "SSE_UNAUTHORIZED", Message: err.Error()},
		})
		return
	}

	if h.redis == nil {
		c.AbortWithStatusJSON(http.StatusServiceUnavailable, errResponse{
			Success: false,
			Error:   &errBody{Code: "SSE_REDIS_UNAVAILABLE", Message: "Redis not configured"},
		})
		return
	}

	// SSE response headers.
	c.Header("Content-Type", "text/event-stream")
	c.Header("Cache-Control", "no-cache")
	c.Header("Connection", "keep-alive")
	c.Header("X-Accel-Buffering", "no") // Disable nginx buffering.

	// Per-connection Redis subscription to the user's notify channel.
	// go-redis PubSub is goroutine-safe per instance; each Subscribe call
	// creates an independent PubSub that is closed on connection end.
	channel := redisclient.NotifyChannel(userID)
	pubsub := h.redis.Redis().Subscribe(c.Request.Context(), channel)
	defer pubsub.Close()

	msgCh := pubsub.Channel()
	ticker := time.NewTicker(heartbeatInterval)
	defer ticker.Stop()

	observability.Logger().Info("sse connection opened",
		"user_id", userID, "channel", channel)

	c.Stream(func(w io.Writer) bool {
		select {
		case <-c.Request.Context().Done():
			return false
		case msg, ok := <-msgCh:
			if !ok {
				// Redis subscription closed (e.g. Redis connection lost).
				observability.Logger().Warn("sse redis subscription closed",
					"user_id", userID)
				return false
			}
			// Forward Redis payload as SSE data frame: `data: {json}\n\n`
			fmt.Fprintf(w, "data: %s\n\n", msg.Payload)
			h.metrics.IncPushed("sse.notification", "delivered")
			return true
		case <-ticker.C:
			// SSE comment frame keeps the connection alive through proxies.
			fmt.Fprintf(w, ": heartbeat\n\n")
			return true
		}
	})

	observability.Logger().Info("sse connection closed", "user_id", userID)
}

// authenticate validates the JWT token. Production uses RS256 via shared-go/jwks;
// DevMode accepts the literal "dev-token" string. The token may be passed via
// the ?token= query parameter or the Authorization: Bearer header.
func (h *Handler) authenticate(c *gin.Context) (string, error) {
	tokenStr := c.Query("token")
	if tokenStr == "" {
		if auth := c.GetHeader("Authorization"); strings.HasPrefix(auth, "Bearer ") {
			tokenStr = strings.TrimPrefix(auth, "Bearer ")
		}
	}
	if tokenStr == "" {
		return "", errMissingToken
	}

	// DevMode shortcut: dev-token maps to a synthetic local user.
	if h.cfg.DevMode && tokenStr == "dev-token" {
		return "dev-user", nil
	}

	// RS256 via shared-go/jwks (preferred, requires iam /.well-known/jwks.json).
	if h.jwksFetcher != nil {
		claims, err := h.jwksFetcher.ValidateToken(tokenStr)
		if err != nil {
			return "", errInvalidToken
		}
		if claims.UserID == "" {
			return "", errMissingUserID
		}
		return claims.UserID, nil
	}

	// DevMode without JWKS: fall back to HS256 with the configured secret.
	if h.cfg.DevMode && h.cfg.JWTSecret != "" {
		return validateHS256(tokenStr, h.cfg.JWTSecret)
	}
	return "", errInvalidToken
}

// validateHS256 parses and validates an HS256 JWT signed with secret. Used
// only in DevMode when no JWKS endpoint is configured (local integration
// tests). Returns the user_id claim on success.
func validateHS256(tokenStr, secret string) (string, error) {
	claims := &jwks.Claims{}
	parsed, err := jwt.ParseWithClaims(tokenStr, claims, func(t *jwt.Token) (any, error) {
		if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok {
			return nil, fmt.Errorf("sse: unexpected signing method %v", t.Header["alg"])
		}
		return []byte(secret), nil
	})
	if err != nil {
		return "", errInvalidToken
	}
	if !parsed.Valid {
		return "", errInvalidToken
	}
	if claims.UserID == "" {
		return "", errMissingUserID
	}
	return claims.UserID, nil
}