xiner/Edu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(push-gateway,msg): redis pubsub backplane for real-time notifications

Browse Source 
M7: ADR-040 Redis Pub/Sub as state routing backplane

- push-gateway: remove Kafka consumer, add SSE endpoint

- SSE: subscribe to Redis user:{userId}:notify on connect

- msg: publish notifications to Redis Pub/Sub instead of HTTP push

- docker-compose: remove Kafka env from push-gateway
This commit is contained in:
SpecialX
2026-07-15 01:28:55 +08:00
parent a75527be80
commit 6af1aa0d82
14 changed files with 402 additions and 484 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
services/push-gateway/internal/config/config.go
View File

@@ -2,9 +2,11 @@
// variables. Required variables (JWT_SECRET in production, PUSH_INTERNAL_TOKEN
// or INTERNAL_API_TOKEN when not in DevMode) cause a fatal exit when missing.
//
// v2 alignment (ARB-013 / msg contract): Kafka topic default changed to
// edu.notify.notification.sent; PUSH_INTERNAL_TOKEN is the canonical env var
// for /internal/* auth (INTERNAL_API_TOKEN kept as a backward-compat alias).
// M7 (ADR-040): Kafka config removed — push-gateway no longer mounts Kafka
// directly. Real-time push is routed through Redis Pub/Sub backplane
// (channel: user:{userId}:notify). PUSH_INTERNAL_TOKEN is the canonical env
// var for /internal/* auth (INTERNAL_API_TOKEN kept as a backward-compat
// alias).
package config
import (
@@ -29,10 +31,6 @@ type Config struct {
HeartbeatInterval int // seconds
// JWT RS256 (via shared-go/jwks, iam /.well-known/jwks.json)
JWKSURL string
// Kafka
KafkaBrokers []string
KafkaNotificationTopic string
KafkaConsumerGroup string
// Instance identity (for Redis SET membership)
InstanceID string
}
@@ -80,11 +78,6 @@ func Load() *Config {
MaxConnsPerUser: env.GetInt("MAX_CONNS_PER_USER", 5),
HeartbeatInterval: env.GetInt("HEARTBEAT_INTERVAL_SECONDS", 30),
JWKSURL: env.Get("JWKS_URL", "http://localhost:50052/.well-known/jwks.json"),
KafkaBrokers: parseBrokers(env.Get("KAFKA_BROKERS", "localhost:9092")),
// ARB-013 canonical topic name: edu.<domain>.<aggregate>.<action>.
// msg's Outbox publisher emits to this topic.
KafkaNotificationTopic: env.Get("KAFKA_NOTIFICATION_TOPIC", "edu.notify.notification.sent"),
KafkaConsumerGroup: env.Get("KAFKA_CONSUMER_GROUP", "push-gateway"),
InstanceID: env.Get("INSTANCE_ID", generateInstanceID()),
}
}
@@ -105,18 +98,6 @@ func parseOrigins(raw string) []string {
return out
}
// parseBrokers splits a comma-separated list of Kafka broker addresses.
func parseBrokers(raw string) []string {
parts := strings.Split(raw, ",")
out := make([]string, 0, len(parts))
for _, p := range parts {
if trimmed := strings.TrimSpace(p); trimmed != "" {
out = append(out, trimmed)
}
}
return out
}
// generateInstanceID returns a stable process-unique identifier. Falls back to
// the hostname when INSTANCE_ID is not explicitly set, allowing each replica to
// be uniquely identifiable in the Redis online-presence SET.

18
services/push-gateway/internal/config/config_test.go
View File

@@ -32,13 +32,6 @@ func TestLoadDevMode(t *testing.T) {
if cfg.HeartbeatInterval != 30 {
t.Errorf("HeartbeatInterval = %d, want 30", cfg.HeartbeatInterval)
}
// ARB-013: canonical topic name is edu.notify.notification.sent.
if cfg.KafkaNotificationTopic != "edu.notify.notification.sent" {
t.Errorf("KafkaNotificationTopic = %q, want edu.notify.notification.sent", cfg.KafkaNotificationTopic)
}
if cfg.KafkaConsumerGroup != "push-gateway" {
t.Errorf("KafkaConsumerGroup = %q, want push-gateway", cfg.KafkaConsumerGroup)
}
if cfg.InstanceID == "" {
t.Error("InstanceID should default to hostname, got empty")
}
@@ -113,17 +106,6 @@ func TestParseOrigins(t *testing.T) {
}
}
// TestParseBrokers verifies comma-separated broker parsing.
func TestParseBrokers(t *testing.T) {
got := parseBrokers("kafka1:9092,kafka2:9092,kafka3:9092")
if len(got) != 3 {
t.Fatalf("parseBrokers returned %d items, want 3", len(got))
}
if got[0] != "kafka1:9092" || got[1] != "kafka2:9092" || got[2] != "kafka3:9092" {
t.Errorf("parseBrokers = %v", got)
}
}
// TestGenerateInstanceID verifies fallback to hostname.
func TestGenerateInstanceID(t *testing.T) {
id := generateInstanceID()

76
services/push-gateway/internal/health/readyz.go
View File

@@ -6,18 +6,19 @@
//
// /readyz reports readiness based on downstream dependency health. Per
// ARB-015 §17.4 / ISSUE-058 the readiness probe uses SOFT FAILURE semantics:
// when Redis or Kafka is unavailable the endpoint still returns HTTP 200 but
// carries `degraded: true` plus a `dependencies` block describing which
// component failed. This prevents Kubernetes from evicting the pod when a
// transient dependency blip occurs, at the cost of accepting some degraded
// behavior (no cross-instance fanout, no Kafka consumption) during the blip.
// Only the Hub being in a shutting-down state returns a non-200 (503).
// when Redis is unavailable the endpoint still returns HTTP 200 but carries
// `degraded: true` plus a `dependencies` block describing which component
// failed. This prevents Kubernetes from evicting the pod when a transient
// dependency blip occurs, at the cost of accepting some degraded behavior
// (no Redis Pub/Sub fanout, no SSE delivery) during the blip. Only the Hub
// being in a shutting-down state returns a non-200 (503).
//
// M7 (ADR-040): Kafka probe removed — push-gateway no longer mounts Kafka.
//
// The hard-failure case is limited to:
// - Hub.closing == true (process is shutting down) -> 503
// - Internal misconfiguration (both Redis and Kafka missing in non-DevMode)
// -> still 200 + degraded, since the pod can still serve local WebSocket
// traffic.
// - Redis missing in non-DevMode -> still 200 + degraded, since the pod can
// still serve local WebSocket traffic.
package health
import (
@@ -26,7 +27,6 @@ import (
"time"
"github.com/edu-cloud/push-gateway/internal/hub"
"github.com/edu-cloud/push-gateway/internal/kafkaconsumer"
"github.com/edu-cloud/push-gateway/internal/redisclient"
"github.com/gin-gonic/gin"
)
@@ -47,12 +47,12 @@ type dependencyStatus struct {
// readyzResponse is the ActionState-shaped envelope returned by /readyz.
// `degraded` is true when at least one non-critical dependency is unhealthy.
type readyzResponse struct {
Status string `json:"status"`
Service string `json:"service"`
InstanceID string `json:"instance_id"`
Degraded bool `json:"degraded"`
Connections int `json:"connections"`
Users int `json:"users"`
Status string `json:"status"`
Service string `json:"service"`
InstanceID string `json:"instance_id"`
Degraded bool `json:"degraded"`
Connections int `json:"connections"`
Users int `json:"users"`
Dependencies map[string]*dependencyStatus `json:"dependencies"`
}
@@ -68,28 +68,26 @@ func Healthz(service string) gin.HandlerFunc {
}
}
// Readyzer builds the /readyz handler. It probes Redis (PING) and Kafka
// (reader lag / connectivity) and reports degraded state per ARB-015 §17.4.
// Readyzer builds the /readyz handler. It probes Redis (PING) and reports
// degraded state per ARB-015 §17.4.
//
// The Hub is used to report local connection counts and to detect the
// shutting-down state (which triggers a hard 503). The redisClient may be nil
// in DevMode; the kafkaConsumer may be nil when KAFKA_BROKERS is unset. Both
// nil cases are reported as degraded rather than failing the probe.
// in DevMode; the nil case is reported as degraded rather than failing the
// probe.
type Readyzer struct {
hub *hub.Hub
redis *redisclient.Client
kafka *kafkaconsumer.Consumer
instance string
service string
}
// NewReadyzer constructs a Readyzer. redis and kafka may be nil; the resulting
// probe will mark the missing dependency as degraded.
func NewReadyzer(h *hub.Hub, r *redisclient.Client, k *kafkaconsumer.Consumer, service, instanceID string) *Readyzer {
// NewReadyzer constructs a Readyzer. redis may be nil; the resulting probe
// will mark the missing dependency as degraded.
func NewReadyzer(h *hub.Hub, r *redisclient.Client, service, instanceID string) *Readyzer {
return &Readyzer{
hub: h,
redis: r,
kafka: k,
instance: instanceID,
service: service,
}
@@ -98,8 +96,6 @@ func NewReadyzer(h *hub.Hub, r *redisclient.Client, k *kafkaconsumer.Consumer, s
// Handler is the gin.HandlerFunc for GET /readyz.
func (rz *Readyzer) Handler(c *gin.Context) {
// Hard failure: Hub is shutting down — reject new traffic.
// (Hub.CloseAll sets closing=true; we treat this as 503 so the load
// balancer stops sending WebSocket upgrades during drain.)
if rz.hub.IsClosing() {
c.JSON(http.StatusServiceUnavailable, readyzResponse{
Status: "shutting_down",
@@ -115,7 +111,7 @@ func (rz *Readyzer) Handler(c *gin.Context) {
return
}
deps := make(map[string]*dependencyStatus, 2)
deps := make(map[string]*dependencyStatus, 1)
degraded := false
// Redis probe (soft failure).
@@ -129,17 +125,6 @@ func (rz *Readyzer) Handler(c *gin.Context) {
}
}
// Kafka probe (soft failure).
if rz.kafka == nil {
deps["kafka"] = &dependencyStatus{Ok: false, Error: "not configured"}
degraded = true
} else {
deps["kafka"] = probeKafka(rz.kafka)
if !deps["kafka"].Ok {
degraded = true
}
}
// Always 200 (unless shutting down) per ISSUE-058/006 soft-failure rule.
c.JSON(http.StatusOK, readyzResponse{
Status: statusText(degraded),
@@ -171,16 +156,3 @@ func probeRedis(r *redisclient.Client) *dependencyStatus {
}
return &dependencyStatus{Ok: true, Latency: time.Since(start).Milliseconds()}
}
// probeKafka checks that the consumer reader is still reachable. We use a
// lightweight Lag() call (segmentio/kafka-go Client API); on failure the
// consumer is marked degraded. Note: a degraded Kafka does NOT block WebSocket
// traffic — it only pauses notification consumption until recovery.
func probeKafka(k *kafkaconsumer.Consumer) *dependencyStatus {
ctx, cancel := context.WithTimeout(context.Background(), probeTimeout)
defer cancel()
if err := k.HealthCheck(ctx); err != nil {
return &dependencyStatus{Ok: false, Error: err.Error()}
}
return &dependencyStatus{Ok: true}
}

253
services/push-gateway/internal/kafkaconsumer/consumer.go
View File

@@ -1,253 +0,0 @@
// Package kafkaconsumer subscribes to the edu.notify.notification.sent topic
// (ARB-013 canonical topic name) on behalf of push-gateway. Each consumed
// NotificationRequested event is dispatched to the Hub for delivery to online
// clients. At-least-once delivery is enforced via manual commit; idempotency
// is provided by Redis SETNX on event_id (TTL 24h).
//
// Topic naming follows ARB-013 (edu.<domain>.<aggregate>.<action>); the
// previous name edu.notification.requested is deprecated (v2 alignment).
package kafkaconsumer
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"time"
"github.com/edu-cloud/push-gateway/internal/hub"
"github.com/edu-cloud/push-gateway/internal/observability"
"github.com/edu-cloud/push-gateway/internal/redisclient"
"github.com/segmentio/kafka-go"
)
// MaxRetries is the per-message retry count before sending to the dead-letter
// topic. After MaxRetries failures the message is committed (skipped) and
// logged; a future P6 enhancement can republish to a DLQ topic.
const MaxRetries = 3
// NotificationRequested is the message shape produced by msg (ai10). Field
// names mirror the JSON wire format produced by msg's Outbox publisher.
type NotificationRequested struct {
EventID string `json:"event_id"`
UserID string `json:"user_id"`
EventType string `json:"event_type"`
Channel string `json:"channel"`
Title string `json:"title"`
Content string `json:"content"`
Data json.RawMessage `json:"data"`
Broadcast bool `json:"broadcast"`
OccurredAt int64 `json:"occurred_at"`
}
// Consumer wraps a kafka.Reader for the notification topic.
type Consumer struct {
reader *kafka.Reader
hub *hub.Hub
redis *redisclient.Client
metrics *observability.Metrics
topic string
groupID string
dlqTopic string
}
// Config configures the Kafka consumer.
type Config struct {
Brokers []string
Topic string
GroupID string
DLQTopic string
}
// New creates a Consumer for the configured topic and group.
func New(cfg Config, h *hub.Hub, r *redisclient.Client, m *observability.Metrics) *Consumer {
if cfg.DLQTopic == "" {
cfg.DLQTopic = cfg.Topic + ".dlq"
}
reader := kafka.NewReader(kafka.ReaderConfig{
Brokers: cfg.Brokers,
Topic: cfg.Topic,
GroupID: cfg.GroupID,
MinBytes: 1,
MaxBytes: 10 * 1024 * 1024,
CommitInterval: 1 * time.Second, // periodic background commit
StartOffset: kafka.LastOffset, // skip historical backlog on first start
})
return &Consumer{
reader: reader,
hub: h,
redis: r,
metrics: m,
topic: cfg.Topic,
groupID: cfg.GroupID,
dlqTopic: cfg.DLQTopic,
}
}
// Run blocks until ctx is canceled, consuming messages and dispatching them.
// Errors are logged but do not stop the consumer unless ctx is canceled.
func (c *Consumer) Run(ctx context.Context) error {
observability.Logger().Info("kafka consumer starting",
"topic", c.topic, "group", c.groupID, "brokers", c.reader.Config().Brokers)
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
msg, err := c.reader.ReadMessage(ctx)
if err != nil {
if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
return nil
}
observability.Logger().Warn("kafka read failed", "err", err)
select {
case <-ctx.Done():
return nil
case <-time.After(2 * time.Second):
}
continue
}
c.metrics.IncKafkaConsumed(c.topic, fmt.Sprintf("%d", msg.Partition))
c.processMessage(ctx, msg)
}
}
// Close releases the underlying reader. Should be called after Run returns.
func (c *Consumer) Close() error {
return c.reader.Close()
}
// HealthCheck verifies that the Kafka broker is reachable. Used by /readyz as
// a soft-failure probe: a failing check marks the consumer as degraded but
// does not return 503 (per ARB-015 §17.4 / ISSUE-058).
//
// The check issues a metadata request for the consumer topic; on success the
// broker is considered reachable. This is cheaper than a Lag() call which
// requires partition assignment to have completed.
func (c *Consumer) HealthCheck(ctx context.Context) error {
conn, err := kafka.DialContext(ctx, "tcp", c.reader.Config().Brokers[0])
if err != nil {
return fmt.Errorf("kafka: dial broker: %w", err)
}
defer conn.Close()
partitions, err := conn.ReadPartitions(c.topic)
if err != nil {
return fmt.Errorf("kafka: read partitions: %w", err)
}
if len(partitions) == 0 {
return fmt.Errorf("kafka: topic %q has no partitions", c.topic)
}
return nil
}
// Topic returns the configured topic name. Exposed for diagnostics.
func (c *Consumer) Topic() string {
return c.topic
}
// GroupID returns the configured consumer group name.
func (c *Consumer) GroupID() string {
return c.groupID
}
// processMessage handles a single Kafka message with retry and idempotency.
func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) {
var event NotificationRequested
if err := json.Unmarshal(msg.Value, &event); err != nil {
observability.Logger().Warn("kafka: invalid message payload",
"topic", c.topic, "partition", msg.Partition, "offset", msg.Offset, "err", err)
c.metrics.IncPushed("notification.invalid", "invalid")
// Commit to skip the poison message.
_ = c.reader.CommitMessages(ctx, msg)
return
}
// eventLabel is the metrics label for IncPushed. Prefer event_type (the
// actual event name like "ExamExtended"); fall back to a generic label.
eventLabel := event.EventType
if eventLabel == "" {
eventLabel = "notification.sent"
}
// Idempotency: skip already-processed events by event_id.
fresh, err := c.redis.DedupEventId(ctx, event.EventID)
if err != nil {
observability.Logger().Warn("kafka: dedup check failed, processing anyway",
"event_id", event.EventID, "err", err)
}
if !fresh {
observability.Logger().Debug("kafka: duplicate event skipped",
"event_id", event.EventID)
c.metrics.IncPushed(eventLabel, "duplicate")
_ = c.reader.CommitMessages(ctx, msg)
return
}
// Retry up to MaxRetries on dispatch failure.
for attempt := 1; attempt <= MaxRetries; attempt++ {
if err := c.dispatch(ctx, event); err == nil {
c.metrics.IncPushed(eventLabel, "delivered")
_ = c.reader.CommitMessages(ctx, msg)
return
} else if attempt < MaxRetries {
observability.Logger().Warn("kafka: dispatch retry",
"event_id", event.EventID, "attempt", attempt, "err", err)
time.Sleep(time.Duration(attempt) * time.Second)
}
}
// MaxRetries exhausted: send to DLQ (best-effort) and commit.
observability.Logger().Error("kafka: dispatch failed after retries, sending to DLQ",
"event_id", event.EventID, "topic", c.topic)
c.sendToDLQ(ctx, msg.Value)
c.metrics.IncPushed(eventLabel, "dlq")
_ = c.reader.CommitMessages(ctx, msg)
}
// dispatch delivers the event to local Hub connections and, when the user is
// not local, asks Redis to fan out via Pub/Sub.
func (c *Consumer) dispatch(ctx context.Context, event NotificationRequested) error {
payload, err := json.Marshal(map[string]any{
"type": "message",
"event": event.EventType,
"data": event.Data,
"timestamp": time.Now().UTC().Format(time.RFC3339),
})
if err != nil {
return fmt.Errorf("marshal payload: %w", err)
}
if event.Broadcast {
// Broadcast: deliver locally + publish to other instances.
c.hub.Broadcast(payload)
return c.redis.PublishBroadcast(ctx, redisclient.CrossInstanceMessage{
Event: event.EventType,
Data: event.Data,
})
}
// Directed: try local first, fall back to Redis Pub/Sub.
if c.hub.HasUser(event.UserID) {
if c.hub.SendToUser(event.UserID, payload) > 0 {
return nil
}
}
return c.redis.PublishUser(ctx, event.UserID, redisclient.CrossInstanceMessage{
Event: event.EventType,
Data: event.Data,
})
}
// sendToDLQ publishes a raw message to the dead-letter topic. Best-effort;
// failure is logged but does not block commit.
func (c *Consumer) sendToDLQ(ctx context.Context, value []byte) {
conn, err := kafka.DialLeader(ctx, "tcp", c.reader.Config().Brokers[0], c.dlqTopic, 0)
if err != nil {
observability.Logger().Error("kafka: DLQ dial failed", "topic", c.dlqTopic, "err", err)
return
}
defer conn.Close()
_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
_, _ = conn.WriteMessages(kafka.Message{Value: value})
}

10
services/push-gateway/internal/redisclient/pubsub.go
View File

@@ -359,6 +359,16 @@ func idempotentKey(eventID string) string {
return "edu:push:idempotent:" + eventID
}
// NotifyChannel returns the Redis Pub/Sub channel name for a user's real-time
// notification stream. M7 (ADR-040): msg service publishes notifications to
// this channel; realtime-gateway instances subscribe per-connection when a
// user connects via SSE.
//
// Channel naming: `user:{userId}:notify`
func NotifyChannel(userID string) string {
return "user:" + userID + ":notify"
}
// ParseUserChannel extracts the userID from a "edu:push:channel:user:<userID>"
// channel name. Returns empty string when the name does not match.
func ParseUserChannel(channel string) string {

217
services/push-gateway/internal/sse/handler.go Normal file
View File

@@ -0,0 +1,217 @@
// 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
}