chore: merge feat/api-gateway-ai01 review docs into main

services/push-gateway/go.mod

@@ -3,14 +3,19 @@ module github.com/edu-cloud/push-gateway
 go 1.25.0
 
 require (
+	github.com/edu-cloud/shared-go v0.0.0
 	github.com/gin-gonic/gin v1.12.0
 	github.com/golang-jwt/jwt/v5 v5.2.1
+	github.com/google/uuid v1.6.0
 	github.com/gorilla/websocket v1.5.3
 	github.com/prometheus/client_golang v1.23.2
+	github.com/redis/go-redis/v9 v9.7.0
+	github.com/segmentio/kafka-go v0.4.48
 	go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.69.0
 	go.opentelemetry.io/otel v1.44.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0
 	go.opentelemetry.io/otel/sdk v1.44.0
+	go.uber.org/zap v1.27.0
 )
 
 require (
@@ -21,6 +26,7 @@ require (
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.7 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/gabriel-vasile/mimetype v1.4.13 // indirect
 	github.com/gin-contrib/sse v1.1.1 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
@@ -30,7 +36,6 @@ require (
 	github.com/go-playground/validator/v10 v10.30.2 // indirect
 	github.com/goccy/go-json v0.10.6 // indirect
 	github.com/goccy/go-yaml v1.19.2 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
@@ -64,3 +69,5 @@ require (
 	google.golang.org/grpc v1.81.1 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 )
+
+replace github.com/edu-cloud/shared-go => ../../packages/shared-go

services/push-gateway/internal/config/config.go

@@ -1,28 +1,50 @@
+// Package config loads runtime configuration for push-gateway from environment
+// variables. Required variables (JWT_SECRET in production, INTERNAL_API_TOKEN
+// when not in DevMode) cause a fatal exit when missing.
 package config
 
 import (
 	"log"
 	"os"
+	"strings"
+
+	"github.com/edu-cloud/shared-go/env"
 )
 
-// Config 持有 push-gateway 运行时配置
+// Config holds all push-gateway runtime configuration.
 type Config struct {
-	Port           string
-	JWTSecret      string
-	DevMode        bool
-	RedisURL       string
-	OTLPEndpoint   string
-	InternalAPIKey string
+	Port             string
+	JWTSecret        string
+	DevMode          bool
+	RedisURL         string
+	OTLPEndpoint     string
+	InternalAPIToken string
+	// WebSocket
+	AllowedOrigins    []string
+	MaxConnsPerUser   int
+	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
 }
 
-// devJWTSecret 是 DevMode 下的默认 JWT 密钥（仅用于本地联调，生产必须配置 JWT_SECRET）
+// devJWTSecret is the default JWT secret used only in DevMode for local
+// development. Production deployments MUST set JWT_SECRET.
 const devJWTSecret = "p1-dev-secret-change-in-production"
 
-// Load 从环境变量加载配置。
-// 非 DevMode 下若 JWT_SECRET 未配置则 fatal 退出；DevMode 下使用默认密钥并打印 warning。
+// Load reads configuration from environment variables.
+//
+// In DevMode (DEV_MODE=true) JWT_SECRET defaults to a dev value and internal
+// API token validation is skipped; in production an unset JWT_SECRET or
+// INTERNAL_API_TOKEN is fatal.
 func Load() *Config {
-	devMode := getEnv("DEV_MODE", "false") == "true"
-	jwtSecret := getEnv("JWT_SECRET", "")
+	devMode := env.GetBool("DEV_MODE", false)
+	jwtSecret := env.Get("JWT_SECRET", "")
 	if jwtSecret == "" {
 		if devMode {
 			log.Println("warning: JWT_SECRET not set, using dev default (DEV_MODE=true)")
@@ -32,20 +54,64 @@ func Load() *Config {
 		}
 	}
 
+	internalToken := env.Get("INTERNAL_API_TOKEN", "")
+	if internalToken == "" && !devMode {
+		log.Fatal("INTERNAL_API_TOKEN must be set in non-dev mode")
+	}
+
 	return &Config{
-		Port:           getEnv("PUSH_GATEWAY_PORT", "8081"),
-		JWTSecret:      jwtSecret,
-		DevMode:        devMode,
-		RedisURL:       getEnv("REDIS_URL", ""),
-		OTLPEndpoint:   getEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318"),
-		InternalAPIKey: getEnv("INTERNAL_API_KEY", ""),
+		Port:             env.Get("PUSH_GATEWAY_PORT", "8081"),
+		JWTSecret:        jwtSecret,
+		DevMode:          devMode,
+		RedisURL:         env.Get("REDIS_URL", "redis://localhost:6379/0"),
+		OTLPEndpoint:     env.Get("OTEL_EXPORTER_OTLP_ENDPOINT", "localhost:4318"),
+		InternalAPIToken: internalToken,
+		AllowedOrigins:   parseOrigins(env.Get("WS_ALLOWED_ORIGINS", "")),
+		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")),
+		KafkaNotificationTopic: env.Get("KAFKA_NOTIFICATION_TOPIC", "edu.notification.requested"),
+		KafkaConsumerGroup:   env.Get("KAFKA_CONSUMER_GROUP", "push-gateway"),
+		InstanceID:       env.Get("INSTANCE_ID", generateInstanceID()),
 	}
 }
 
-// getEnv 读取环境变量，缺失时返回 fallback
-func getEnv(key, fallback string) string {
-	if v := os.Getenv(key); v != "" {
-		return v
+// parseOrigins splits a comma-separated list of allowed WebSocket origins.
+// Empty string yields nil (all origins rejected in production).
+func parseOrigins(raw string) []string {
+	if raw == "" {
+		return nil
 	}
-	return fallback
+	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
+}
+
+// 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.
+func generateInstanceID() string {
+	host, err := os.Hostname()
+	if err != nil || host == "" {
+		return "push-gateway-unknown"
+	}
+	return host
 }

services/push-gateway/internal/hub/hub.go

@@ -1,47 +1,151 @@
+// Package hub manages the in-memory WebSocket connection pool for push-gateway.
+//
+// Hub is the single process-wide registry of all live WebSocket connections,
+// indexed by userID -> connID -> *Connection. Each Connection owns a send
+// channel consumed by a dedicated writer goroutine; Hub exposes Register /
+// Unregister / SendToUser / Broadcast operations.
+//
+// The Hub also tracks per-user connection counts to enforce the
+// MaxConnectionsPerUser limit (default 5) and provides CloseAll for graceful
+// shutdown. Cross-instance presence and message fanout are delegated to a
+// RedisClient (see internal/redis), keeping Hub free of any persistence layer.
 package hub
 
 import (
+	"errors"
 	"sync"
+	"time"
 
 	"github.com/gorilla/websocket"
+	"github.com/google/uuid"
 )
 
-// Connection 包装单个 WebSocket 连接及其异步发送通道
+// ErrTooManyConnections is returned by Register when the per-user connection
+// cap has been reached. Callers should translate this into a 429 /
+// PUSH_TOO_MANY_CONNECTIONS response or a close frame with code 1008.
+var ErrTooManyConnections = errors.New("too many connections per user")
+
+// ErrHubClosing is returned by Register when the Hub is shutting down and no
+// new connections are accepted.
+var ErrHubClosing = errors.New("hub is closing")
+
+// sendBufferSize is the per-connection send channel capacity. A full channel
+// triggers message drop with metric increment (see Send).
+const sendBufferSize = 64
+
+// Connection wraps a single WebSocket connection together with its asynchronous
+// send channel and lifecycle metadata.
 type Connection struct {
-	UserID string
-	conn   *websocket.Conn
-	send   chan []byte
+	UserID    string
+	ConnID    string
+	conn      *websocket.Conn
+	send      chan []byte
+	closeOnce sync.Once
+	closed    bool
+	mu        sync.Mutex
+	lastPing  time.Time
 }
 
-// Send 将消息投递到该连接的发送通道（非阻塞，通道满则丢弃）
-func (c *Connection) Send(message []byte) {
+// Send delivers message to the connection's writer goroutine. When the send
+// channel is full the message is dropped (non-blocking) and the caller is
+// notified via the returned false value so it can increment the
+// push_gateway_messages_dropped_total{reason="channel_full"} metric.
+func (c *Connection) Send(message []byte) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return false
+	}
 	select {
 	case c.send <- message:
+		return true
 	default:
-		// 通道满，丢弃消息避免阻塞 hub
+		return false
 	}
 }
 
-// Outgoing 返回该连接的发送通道，供写协程消费
+// Outgoing exposes the send channel for the writer goroutine. The channel is
+// closed by Close, terminating the consumer loop.
 func (c *Connection) Outgoing() <-chan []byte {
 	return c.send
 }
 
-// Hub 管理所有在线 WebSocket 连接，按 userID 索引
-type Hub struct {
-	mu      sync.RWMutex
-	clients map[string]map[*Connection]bool
+// Close shuts down the connection's send channel exactly once. Subsequent
+// Send calls become no-ops.
+func (c *Connection) Close() {
+	c.closeOnce.Do(func() {
+		c.mu.Lock()
+		c.closed = true
+		c.mu.Unlock()
+		close(c.send)
+	})
 }
 
-// NewHub 创建 Hub 实例
-func NewHub() *Hub {
+// LastPing returns the most recent heartbeat timestamp.
+func (c *Connection) LastPing() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.lastPing
+}
+
+// TouchPing updates the last heartbeat timestamp to now.
+func (c *Connection) TouchPing() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.lastPing = time.Now()
+}
+
+// Conn returns the underlying gorilla websocket.Conn for control-frame
+// operations (Ping/Pong/Close). Used by the reader goroutine.
+func (c *Connection) Conn() *websocket.Conn {
+	return c.conn
+}
+
+// Hub is the process-wide registry of live WebSocket connections.
+//
+// clients maps userID -> connID -> *Connection so a single user may hold
+// multiple simultaneous connections (multi-device). counters caches the
+// per-user connection count to make the MaxConnectionsPerUser check O(1).
+type Hub struct {
+	mu       sync.RWMutex
+	clients  map[string]map[string]*Connection
+	counters map[string]int
+	closing  bool
+
+	// maxConnsPerUser caps the number of simultaneous connections per user.
+	maxConnsPerUser int
+
+	// onRegister / onUnregister are optional hooks used by the Redis presence
+	// layer to maintain the cross-instance online SET. They are invoked outside
+	// the Hub lock to avoid blocking the registry on Redis round-trips.
+	onRegister   func(userID, connID string)
+	onUnregister func(userID, connID string)
+}
+
+// NewHub returns a Hub with the given per-user connection cap.
+func NewHub(maxConnsPerUser int) *Hub {
+	if maxConnsPerUser <= 0 {
+		maxConnsPerUser = 5
+	}
 	return &Hub{
-		clients: make(map[string]map[*Connection]bool),
+		clients:         make(map[string]map[string]*Connection),
+		counters:        make(map[string]int),
+		maxConnsPerUser: maxConnsPerUser,
 	}
 }
 
-// ClientCount 返回当前在线连接总数（供 /readyz 探针使用）
-func (h *Hub) ClientCount() int {
+// SetPresenceHooks registers callbacks invoked after Register / Unregister.
+// Used by the Redis presence layer to keep edu:push:online:<userID> in sync.
+func (h *Hub) SetPresenceHooks(onRegister, onUnregister func(userID, connID string)) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.onRegister = onRegister
+	h.onUnregister = onUnregister
+}
+
+// ActiveConnections returns the total number of live connections across all
+// users. Exposed via /healthz and the push_gateway_active_connections gauge.
+func (h *Hub) ActiveConnections() int {
 	h.mu.RLock()
 	defer h.mu.RUnlock()
 	count := 0
@@ -51,60 +155,162 @@ func (h *Hub) ClientCount() int {
 	return count
 }
 
-// Register 注册一个用户连接，返回 Connection 供调用方持有
-func (h *Hub) Register(userID string, conn *websocket.Conn) *Connection {
-	c := &Connection{
-		UserID: userID,
-		conn:   conn,
-		send:   make(chan []byte, 64),
-	}
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	if h.clients[userID] == nil {
-		h.clients[userID] = make(map[*Connection]bool)
-	}
-	h.clients[userID][c] = true
-	return c
+// UserCount returns the number of distinct online users (used by /healthz).
+func (h *Hub) UserCount() int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return len(h.clients)
 }
 
-// Unregister 注销一个用户连接并关闭其发送通道
-func (h *Hub) Unregister(c *Connection) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	conns, ok := h.clients[c.UserID]
-	if !ok {
-		return
-	}
-	if _, exists := conns[c]; exists {
-		delete(conns, c)
-		close(c.send)
-	}
-	if len(conns) == 0 {
-		delete(h.clients, c.UserID)
-	}
-}
-
-// SendToUser 向指定用户的所有在线连接推送消息
-func (h *Hub) SendToUser(userID string, message []byte) error {
+// HasUser reports whether at least one local connection exists for userID.
+// Used by the /internal/push path to decide between local delivery and Redis
+// Pub/Sub fanout.
+func (h *Hub) HasUser(userID string) bool {
 	h.mu.RLock()
 	defer h.mu.RUnlock()
 	conns, ok := h.clients[userID]
-	if !ok {
-		return nil
-	}
-	for c := range conns {
-		c.Send(message)
-	}
-	return nil
+	return ok && len(conns) > 0
 }
 
-// Broadcast 向所有在线连接广播消息
-func (h *Hub) Broadcast(message []byte) {
-	h.mu.RLock()
-	defer h.mu.RUnlock()
-	for _, conns := range h.clients {
-		for c := range conns {
-			c.Send(message)
+// Register adds a new connection for userID. Returns ErrTooManyConnections
+// when the per-user cap is exceeded and ErrHubClosing when the Hub is shutting
+// down. The returned Connection must be Unregistered by the caller on close.
+func (h *Hub) Register(userID string, conn *websocket.Conn) (*Connection, error) {
+	h.mu.Lock()
+	if h.closing {
+		h.mu.Unlock()
+		return nil, ErrHubClosing
+	}
+	if h.counters[userID] >= h.maxConnsPerUser {
+		h.mu.Unlock()
+		return nil, ErrTooManyConnections
+	}
+	connID := uuid.NewString()
+	c := &Connection{
+		UserID:   userID,
+		ConnID:   connID,
+		conn:     conn,
+		send:     make(chan []byte, sendBufferSize),
+		lastPing: time.Now(),
+	}
+	if h.clients[userID] == nil {
+		h.clients[userID] = make(map[string]*Connection)
+	}
+	h.clients[userID][connID] = c
+	h.counters[userID]++
+	onRegister := h.onRegister
+	h.mu.Unlock()
+
+	if onRegister != nil {
+		onRegister(userID, connID)
+	}
+	return c, nil
+}
+
+// Unregister removes a connection and closes its send channel. Safe to call
+// multiple times for the same Connection.
+func (h *Hub) Unregister(c *Connection) {
+	h.mu.Lock()
+	conns, ok := h.clients[c.UserID]
+	if !ok {
+		h.mu.Unlock()
+		return
+	}
+	if _, exists := conns[c.ConnID]; exists {
+		delete(conns, c.ConnID)
+		h.counters[c.UserID]--
+		if h.counters[c.UserID] <= 0 {
+			delete(h.counters, c.UserID)
+		}
+		if len(conns) == 0 {
+			delete(h.clients, c.UserID)
 		}
 	}
+	onUnregister := h.onUnregister
+	h.mu.Unlock()
+
+	c.Close()
+	if onUnregister != nil {
+		onUnregister(c.UserID, c.ConnID)
+	}
+}
+
+// SendToUser delivers message to every local connection of userID. Returns
+// the number of connections the message was successfully queued to. A zero
+// return value signals "user not online locally" so the caller can fall back
+// to Redis Pub/Sub fanout.
+func (h *Hub) SendToUser(userID string, message []byte) int {
+	h.mu.RLock()
+	conns := h.clients[userID]
+	// Snapshot the slice under the read lock to avoid holding it during Send.
+	snapshot := make([]*Connection, 0, len(conns))
+	for _, c := range conns {
+		snapshot = append(snapshot, c)
+	}
+	h.mu.RUnlock()
+
+	delivered := 0
+	for _, c := range snapshot {
+		if c.Send(message) {
+			delivered++
+		}
+	}
+	return delivered
+}
+
+// Broadcast delivers message to every live connection in this instance.
+// Cross-instance broadcast is handled separately via Redis Pub/Sub.
+func (h *Hub) Broadcast(message []byte) int {
+	h.mu.RLock()
+	snapshot := make([]*Connection, 0, h.ActiveConnections())
+	for _, conns := range h.clients {
+		for _, c := range conns {
+			snapshot = append(snapshot, c)
+		}
+	}
+	h.mu.RUnlock()
+
+	delivered := 0
+	for _, c := range snapshot {
+		if c.Send(message) {
+			delivered++
+		}
+	}
+	return delivered
+}
+
+// CloseAll marks the Hub as closing (rejecting further Register calls) and
+// sends a close frame to every live connection. Used during graceful shutdown.
+// The caller should wait for connections to drain after CloseAll returns.
+func (h *Hub) CloseAll() {
+	h.mu.Lock()
+	h.closing = true
+	all := make([]*Connection, 0, h.ActiveConnections())
+	for _, conns := range h.clients {
+		for _, c := range conns {
+			all = append(all, c)
+		}
+	}
+	h.mu.Unlock()
+
+	for _, c := range all {
+		// Send a graceful close frame (1001 going away) and close the send
+		// channel so the writer goroutine exits.
+		_ = c.conn.WriteControl(
+			websocket.CloseMessage,
+			websocket.FormatCloseMessage(websocket.CloseGoingAway, "server shutting down"),
+			time.Now().Add(2*time.Second),
+		)
+		c.Close()
+	}
+}
+
+// ForEachUser iterates over all locally online userIDs. Used by the Redis
+// presence layer to rebuild the online SET on startup (ISSUE-058).
+func (h *Hub) ForEachUser(fn func(userID string)) {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	for userID := range h.clients {
+		fn(userID)
+	}
 }

services/push-gateway/internal/kafkaconsumer/consumer.go

@@ -0,0 +1,214 @@
+// Package kafkaconsumer subscribes to the edu.notification.requested topic
+// (ISSUE-053 final 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 the G16 rule (edu.<domain>.<aggregate>.<action>); the
+// previous abstract name edu.notification.events / edu.msg.notification.events
+// is deprecated (see president-final-rulings §1.5).
+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()
+}
+
+// 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.requested", "invalid")
+		// Commit to skip the poison message.
+		_ = c.reader.CommitMessages(ctx, msg)
+		return
+	}
+
+	// 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("notification.requested", "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("notification.requested", "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("notification.requested", "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})
+}

services/push-gateway/internal/observability/logger.go

@@ -0,0 +1,77 @@
+// Package observability provides slog-based structured logging for push-gateway.
+//
+// The logger emits JSON in production and a human-readable console format in
+// development (DEV_MODE=true). Trace correlation is handled by attaching a
+// trace_id attribute from the OpenTelemetry span context when present.
+package observability
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"time"
+
+	"go.opentelemetry.io/otel/trace"
+)
+
+// ctxKey is an unexported type so context values cannot collide with callers.
+type ctxKey struct{}
+
+// logger is the process-wide *slog.Logger initialized by InitLogger.
+var logger *slog.Logger
+
+// InitLogger configures the global slog logger. In DevMode a console handler
+// with colored levels is used; otherwise a JSON handler at Info level is used.
+// The logger is tagged with service=push-gateway on every emitted record.
+func InitLogger(devMode bool) *slog.Logger {
+	var handler slog.Handler
+	opts := &slog.HandlerOptions{
+		Level: slog.LevelInfo,
+	}
+	if devMode {
+		opts.Level = slog.LevelDebug
+		handler = slog.NewTextHandler(os.Stdout, opts)
+	} else {
+		handler = slog.NewJSONHandler(os.Stdout, opts)
+	}
+	logger = slog.New(handler).With("service", "push-gateway")
+	slog.SetDefault(logger)
+	return logger
+}
+
+// Logger returns the process-wide logger initialized by InitLogger. When Init
+// has not been called a fallback JSON logger is returned so callers never
+// receive a nil logger.
+func Logger() *slog.Logger {
+	if logger == nil {
+		logger = slog.New(slog.NewJSONHandler(os.Stdout, nil)).With("service", "push-gateway")
+	}
+	return logger
+}
+
+// WithContext returns a copy of ctx carrying l so it can later be retrieved
+// via FromContext.
+func WithContext(ctx context.Context, l *slog.Logger) context.Context {
+	return context.WithValue(ctx, ctxKey{}, l)
+}
+
+// FromContext returns the logger stored in ctx. When an OpenTelemetry span is
+// active the returned logger is decorated with a trace_id field so log lines
+// can be correlated to traces. If no logger was stored in ctx the
+// process-wide logger is returned (never nil).
+func FromContext(ctx context.Context) *slog.Logger {
+	l, ok := ctx.Value(ctxKey{}).(*slog.Logger)
+	if !ok || l == nil {
+		l = Logger()
+	}
+	if sc := trace.SpanContextFromContext(ctx); sc.HasTraceID() {
+		return l.With("trace_id", sc.TraceID().String())
+	}
+	return l
+}
+
+// Now returns the current UTC time in RFC3339 format. Convenience helper for
+// log field construction.
+func Now() string {
+	return time.Now().UTC().Format(time.RFC3339)
+}

services/push-gateway/internal/observability/metrics.go

@@ -0,0 +1,124 @@
+// Package observability also exposes Prometheus metrics for push-gateway.
+//
+// The metrics cover the full surface area described in 02 §6.4: active
+// connections, messages pushed/dropped, heartbeats, disconnects, Redis Pub/Sub
+// latency, Kafka consumption and Redis SET rebuilds. All metrics are
+// registered with the global prometheus.DefaultRegisterer on package init.
+package observability
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+// Metrics holds all push-gateway Prometheus metric handles.
+type Metrics struct {
+	ActiveConnections   prometheus.Gauge
+	MessagesPushed      *prometheus.CounterVec
+	MessagesDropped     *prometheus.CounterVec
+	Heartbeats          prometheus.Counter
+	Disconnects         *prometheus.CounterVec
+	RedisPubSubLatency  *prometheus.HistogramVec
+	KafkaConsumed       *prometheus.CounterVec
+	RedisSetRebuild     prometheus.Counter
+	ConnectionsPerUser  prometheus.Gauge
+}
+
+// metrics is the process-wide metric set, initialized once by NewMetrics.
+var metrics *Metrics
+
+// NewMetrics registers and returns the push-gateway metric set. Subsequent
+// calls return the same instance to avoid duplicate-registration panics.
+func NewMetrics() *Metrics {
+	if metrics != nil {
+		return metrics
+	}
+	metrics = &Metrics{
+		ActiveConnections: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: "push_gateway",
+			Name:      "active_connections",
+			Help:      "Current number of live WebSocket connections.",
+		}),
+		MessagesPushed: promauto.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "push_gateway",
+			Name:      "messages_pushed_total",
+			Help:      "Total messages pushed, partitioned by event and result.",
+		}, []string{"event", "result"}),
+		MessagesDropped: promauto.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "push_gateway",
+			Name:      "messages_dropped_total",
+			Help:      "Messages dropped, partitioned by reason.",
+		}, []string{"reason"}),
+		Heartbeats: promauto.NewCounter(prometheus.CounterOpts{
+			Namespace: "push_gateway",
+			Name:      "heartbeat_total",
+			Help:      "Total WebSocket Ping heartbeats received.",
+		}),
+		Disconnects: promauto.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "push_gateway",
+			Name:      "disconnect_total",
+			Help:      "Disconnects, partitioned by reason.",
+		}, []string{"reason"}),
+		RedisPubSubLatency: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: "push_gateway",
+			Name:      "redis_pubsub_latency_seconds",
+			Help:      "Redis Pub/Sub round-trip latency in seconds.",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 12),
+		}, []string{"direction"}),
+		KafkaConsumed: promauto.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "push_gateway",
+			Name:      "kafka_consumed_total",
+			Help:      "Kafka messages consumed, partitioned by topic and partition.",
+		}, []string{"topic", "partition"}),
+		RedisSetRebuild: promauto.NewCounter(prometheus.CounterOpts{
+			Namespace: "push_gateway",
+			Name:      "redis_set_rebuild_total",
+			Help:      "Total Redis online-presence SET rebuilds (startup or recovery).",
+		}),
+		ConnectionsPerUser: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: "push_gateway",
+			Name:      "connections_per_user_max",
+			Help:      "Maximum concurrent connections held by a single user.",
+		}),
+	}
+	return metrics
+}
+
+// Metrics returns the process-wide metric set. When NewMetrics has not been
+// called a new set is registered on demand.
+func MetricsInstance() *Metrics {
+	if metrics == nil {
+		return NewMetrics()
+	}
+	return metrics
+}
+
+// IncMessageDropped is a convenience helper for the channel-full case.
+func (m *Metrics) IncMessageDropped(reason string) {
+	m.MessagesDropped.WithLabelValues(reason).Inc()
+}
+
+// IncPushed is a convenience helper for the push result counter.
+func (m *Metrics) IncPushed(event, result string) {
+	m.MessagesPushed.WithLabelValues(event, result).Inc()
+}
+
+// IncDisconnect is a convenience helper for the disconnect counter.
+func (m *Metrics) IncDisconnect(reason string) {
+	m.Disconnects.WithLabelValues(reason).Inc()
+}
+
+// ObservePubSubLatency records a Redis Pub/Sub round-trip duration.
+func (m *Metrics) ObservePubSubLatency(direction string, seconds float64) {
+	m.RedisPubSubLatency.WithLabelValues(direction).Observe(seconds)
+}
+
+// IncKafkaConsumed increments the Kafka consumed counter.
+func (m *Metrics) IncKafkaConsumed(topic, partition string) {
+	m.KafkaConsumed.WithLabelValues(topic, partition).Inc()
+}
+
+// IncRedisSetRebuild increments the SET rebuild counter (ISSUE-058 metric).
+func (m *Metrics) IncRedisSetRebuild() {
+	m.RedisSetRebuild.Inc()
+}

services/push-gateway/internal/observability/tracer.go

@@ -1,73 +1,35 @@
+// Package observability also bootstraps OpenTelemetry tracing. The tracer is
+// initialized via shared-go/tracer so push-gateway and api-gateway share the
+// same SDK configuration.
 package observability
 
 import (
 	"context"
-	"log"
-	"net/url"
-	"time"
+	"fmt"
+	"os"
 
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
-	"go.opentelemetry.io/otel/propagation"
-	"go.opentelemetry.io/otel/sdk/resource"
-	sdktrace "go.opentelemetry.io/otel/sdk/trace"
-	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
+	"github.com/edu-cloud/shared-go/tracer"
 )
 
-// InitTracer 初始化 OpenTelemetry tracer.
-//
-// endpoint 为 "http://host:port" 格式（如 "http://localhost:4318"）；
-// 为空时跳过初始化（tracing disabled）。
-//
-// 返回 shutdown 函数，应在服务退出时调用以 flush 待发送 span.
+// InitTracer initializes the OTel tracer using shared-go/tracer. When the
+// OTLP endpoint is empty (or "localhost:4318") the exporter still runs but
+// emits no spans if no collector is listening. Returns a shutdown function
+// that must be called on process exit.
 func InitTracer(serviceName, endpoint string) func() {
-	if endpoint == "" {
-		log.Println("OTEL endpoint not set, tracing disabled")
+	// shared-go/tracer reads OTEL_EXPORTER_OTLP_ENDPOINT directly; override it
+	// explicitly so push-gateway's own config value takes precedence.
+	if endpoint != "" {
+		_ = os.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint)
+	}
+	if err := tracer.Init(serviceName); err != nil {
+		// Tracing is best-effort: log and continue without a tracer rather than
+		// failing the whole process.
+		fmt.Fprintf(os.Stderr, "push-gateway: tracer init failed: %v (tracing disabled)\n", err)
 		return func() {}
 	}
-
-	u, err := url.Parse(endpoint)
-	if err != nil || u.Host == "" {
-		log.Printf("invalid OTEL endpoint %q, tracing disabled", endpoint)
-		return func() {}
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	exporter, err := otlptracehttp.New(ctx,
-		otlptracehttp.WithEndpoint(u.Host),
-		otlptracehttp.WithInsecure(),
-	)
-	if err != nil {
-		log.Printf("failed to create OTLP exporter: %v, tracing disabled", err)
-		return func() {}
-	}
-
-	res, err := resource.New(ctx,
-		resource.WithAttributes(semconv.ServiceName(serviceName)),
-	)
-	if err != nil {
-		log.Printf("failed to create resource: %v", err)
-		return func() {}
-	}
-
-	tp := sdktrace.NewTracerProvider(
-		sdktrace.WithBatcher(exporter),
-		sdktrace.WithResource(res),
-	)
-	otel.SetTracerProvider(tp)
-	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
-		propagation.TraceContext{},
-		propagation.Baggage{},
-	))
-
-	log.Printf("OpenTelemetry tracer initialized for %s (endpoint=%s)", serviceName, u.Host)
 	return func() {
-		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		defer cancel()
-		if err := tp.Shutdown(shutdownCtx); err != nil {
-			log.Printf("failed to shutdown tracer: %v", err)
+		if err := tracer.Shutdown(context.Background()); err != nil {
+			fmt.Fprintf(os.Stderr, "push-gateway: tracer shutdown failed: %v\n", err)
 		}
 	}
 }

services/push-gateway/internal/redisclient/pubsub.go

@@ -0,0 +1,370 @@
+// Package redisclient manages cross-instance presence and message fanout
+// using Redis Pub/Sub plus the online-presence SET described in 02 §3.1 and
+// §8.4. It implements the ISSUE-058 startup-rebuild mechanism: when a Hub
+// starts it clears stale members belonging to its own instanceID from the
+// online SET, then re-adds itself for every locally connected user.
+//
+// Keys / channels:
+//
+//	edu:push:online:<userID>           SET (members are instanceID)        TTL 60s
+//	edu:push:session:<userID>:<connID> HASH (instanceID, device, lastPing) TTL 60s
+//	edu:push:channel:user:<userID>     Pub/Sub channel (directed push)
+//	edu:push:channel:broadcast         Pub/Sub channel (broadcast push)
+//	edu:push:idempotent:<event_id>     SETNX key (Kafka dedup)            TTL 24h
+package redisclient
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/edu-cloud/push-gateway/internal/hub"
+	"github.com/edu-cloud/push-gateway/internal/observability"
+	"github.com/redis/go-redis/v9"
+)
+
+// PresenceTTL is the expiry applied to the online-presence SET and per-session
+// HASH. The Hub refreshes it on every heartbeat; if a push-gateway instance
+// crashes the SET entries expire naturally within this window.
+const PresenceTTL = 60 * time.Second
+
+// IdempotentTTL is the dedup window for Kafka event_id (24h).
+const IdempotentTTL = 24 * time.Hour
+
+// CrossInstanceMessage is the payload published to the
+// edu:push:channel:user:<userID> and edu:push:channel:broadcast channels.
+// Subscribers unmarshal it and dispatch to local Hub connections.
+type CrossInstanceMessage struct {
+	Kind     string          `json:"kind"`     // "user" | "broadcast"
+	UserID   string          `json:"userId"`   // set when Kind == "user"
+	Event    string          `json:"event"`
+	Data     json.RawMessage `json:"data"`
+	Origin   string          `json:"origin"`   // instanceID of publisher (loopback suppression)
+	TraceID  string          `json:"traceId,omitempty"`
+}
+
+// Client wraps a redis.Client with push-gateway specific presence and
+// fanout helpers. The Hub is wired via SetHub so the subscriber can dispatch
+// incoming Pub/Sub messages directly to local connections.
+type Client struct {
+	rdb        *redis.Client
+	instanceID string
+	hub        *hub.Hub
+	metrics    *observability.Metrics
+}
+
+// New connects to Redis at url and returns a Client ready for presence and
+// Pub/Sub operations. The instanceID identifies this process in the online
+// SET (see ISSUE-058).
+func New(url, instanceID string, m *observability.Metrics) (*Client, error) {
+	if url == "" {
+		return nil, errors.New("redisclient: empty URL")
+	}
+	opts, err := redis.ParseURL(url)
+	if err != nil {
+		return nil, fmt.Errorf("redisclient: parsing URL %q: %w", url, err)
+	}
+	rdb := redis.NewClient(opts)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	if err := rdb.Ping(ctx).Err(); err != nil {
+		_ = rdb.Close()
+		return nil, fmt.Errorf("redisclient: ping failed: %w", err)
+	}
+	return &Client{
+		rdb:        rdb,
+		instanceID: instanceID,
+		metrics:    m,
+	}, nil
+}
+
+// SetHub wires the Hub so the Pub/Sub subscriber can deliver messages locally.
+// Also installs presence hooks on the Hub so Register/Unregister keep the
+// online SET in sync.
+func (c *Client) SetHub(h *hub.Hub) {
+	c.hub = h
+	h.SetPresenceHooks(c.onRegister, c.onUnregister)
+}
+
+// Redis returns the underlying *redis.Client for direct use by callers that
+// need bespoke commands (e.g. SETNX for Kafka idempotency).
+func (c *Client) Redis() *redis.Client {
+	return c.rdb
+}
+
+// Close releases the Redis connection.
+func (c *Client) Close() error {
+	return c.rdb.Close()
+}
+
+// Ping is a lightweight liveness probe used by /readyz.
+func (c *Client) Ping(ctx context.Context) error {
+	return c.rdb.Ping(ctx).Err()
+}
+
+// ----- Presence SET (ISSUE-058) -------------------------------------------
+
+// RebuildPresenceOnStartup clears stale SET entries belonging to this
+// instanceID and re-adds the instanceID to edu:push:online:<userID> for every
+// locally connected user (ISSUE-058 startup rebuild).
+//
+// This addresses the "ghost member" problem: when a push-gateway instance
+// restarts, its previous instanceID entries may still be present in Redis
+// (within the 60s TTL window). Re-adding the same instanceID would be a
+// no-op, but if the instance previously crashed without removing members we
+// must avoid carrying stale entries that no longer correspond to live
+// connections. We clear first, then rebuild from the in-memory Hub state.
+//
+// 60s inconsistency window: between crash and SET expiry, /internal/push may
+// report online:true for a user whose connection is gone; msg must tolerate
+// delivered:false by retrying or falling back to offline push.
+func (c *Client) RebuildPresenceOnStartup(ctx context.Context) error {
+	if c.hub == nil {
+		return errors.New("redisclient: Hub not set")
+	}
+	start := time.Now()
+	c.metrics.IncRedisSetRebuild()
+
+	// Clear stale members belonging to this instanceID across all online SETs.
+	// We scan keys matching edu:push:online:* and SREM our instanceID from each.
+	iter := c.rdb.Scan(ctx, 0, "edu:push:online:*", 100).Iterator()
+	cleared := 0
+	for iter.Next(ctx) {
+		key := iter.Val()
+		if n, err := c.rdb.SRem(ctx, key, c.instanceID).Result(); err == nil && n > 0 {
+			cleared++
+		}
+	}
+	if err := iter.Err(); err != nil {
+		return fmt.Errorf("redisclient: scanning online SETs: %w", err)
+	}
+
+	// Re-add instanceID for every locally online user.
+	readded := 0
+	c.hub.ForEachUser(func(userID string) {
+		key := onlineKey(userID)
+		pipe := c.rdb.Pipeline()
+		pipe.SAdd(ctx, key, c.instanceID)
+		pipe.Expire(ctx, key, PresenceTTL)
+		if _, err := pipe.Exec(ctx); err == nil {
+			readded++
+		}
+	})
+
+	observability.Logger().Info("redis presence rebuilt",
+		"cleared", cleared,
+		"readded", readded,
+		"duration_ms", time.Since(start).Milliseconds(),
+	)
+	return nil
+}
+
+// onRegister is the Hub hook invoked after a new connection is registered.
+// Adds the instanceID to the user's online SET and refreshes TTL.
+func (c *Client) onRegister(userID, connID string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	key := onlineKey(userID)
+	pipe := c.rdb.Pipeline()
+	pipe.SAdd(ctx, key, c.instanceID)
+	pipe.Expire(ctx, key, PresenceTTL)
+	if _, err := pipe.Exec(ctx); err != nil {
+		observability.Logger().Warn("redis SAdd failed",
+			"user_id", userID, "err", err)
+	}
+}
+
+// onUnregister is the Hub hook invoked after a connection is removed.
+// Removes the instanceID from the SET and deletes the key when empty.
+func (c *Client) onUnregister(userID, connID string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	key := onlineKey(userID)
+	// Only remove if this user has no further local connections (Hub already
+	// removed the connection by the time the hook fires).
+	if c.hub != nil && !c.hub.HasUser(userID) {
+		if err := c.rdb.SRem(ctx, key, c.instanceID).Err(); err != nil {
+			observability.Logger().Warn("redis SRem failed",
+				"user_id", userID, "err", err)
+		}
+		// Best-effort DEL when SET becomes empty.
+		_ = c.rdb.Del(ctx, key).Err()
+	}
+}
+
+// RefreshPresence extends the online SET TTL on heartbeat. Called by the
+// WebSocket reader goroutine when a Ping control frame arrives.
+func (c *Client) RefreshPresence(ctx context.Context, userID string) error {
+	return c.rdb.Expire(ctx, onlineKey(userID), PresenceTTL).Err()
+}
+
+// IsOnline reports whether userID has any instance in the online SET.
+// Returns false when Redis is unreachable; callers should treat the
+// /internal/push response as authoritative for delivery.
+func (c *Client) IsOnline(ctx context.Context, userID string) (bool, []string, error) {
+	members, err := c.rdb.SMembers(ctx, onlineKey(userID)).Result()
+	if err != nil {
+		return false, nil, err
+	}
+	return len(members) > 0, members, nil
+}
+
+// ----- Pub/Sub fanout ------------------------------------------------------
+
+// PublishUser publishes a directed message to edu:push:channel:user:<userID>.
+// All push-gateway instances subscribe to this channel; the one holding the
+// user will deliver locally, others drop it.
+func (c *Client) PublishUser(ctx context.Context, userID string, msg CrossInstanceMessage) error {
+	msg.Kind = "user"
+	msg.UserID = userID
+	msg.Origin = c.instanceID
+	payload, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("redisclient: marshalling user msg: %w", err)
+	}
+	start := time.Now()
+	if err := c.rdb.Publish(ctx, userChannel(userID), payload).Err(); err != nil {
+		return fmt.Errorf("redisclient: PUBLISH user: %w", err)
+	}
+	c.metrics.ObservePubSubLatency("publish", time.Since(start).Seconds())
+	return nil
+}
+
+// PublishBroadcast publishes a broadcast message to all instances.
+func (c *Client) PublishBroadcast(ctx context.Context, msg CrossInstanceMessage) error {
+	msg.Kind = "broadcast"
+	msg.Origin = c.instanceID
+	payload, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("redisclient: marshalling broadcast msg: %w", err)
+	}
+	start := time.Now()
+	if err := c.rdb.Publish(ctx, broadcastChannel(), payload).Err(); err != nil {
+		return fmt.Errorf("redisclient: PUBLISH broadcast: %w", err)
+	}
+	c.metrics.ObservePubSubLatency("publish", time.Since(start).Seconds())
+	return nil
+}
+
+// SubscribeAll subscribes to all per-user channels (pattern) and the
+// broadcast channel, dispatching received messages to the local Hub.
+// The returned cancel function stops the subscriber; it blocks until the
+// goroutine has fully drained.
+func (c *Client) SubscribeAll(ctx context.Context) (cancel func() error, err error) {
+	if c.hub == nil {
+		return nil, errors.New("redisclient: Hub not set")
+	}
+
+	subCtx, subCancel := context.WithCancel(ctx)
+	userPubsub := c.rdb.PSubscribe(subCtx, "edu:push:channel:user:*")
+	broadcastPubsub := c.rdb.Subscribe(subCtx, broadcastChannel())
+
+	go c.dispatchLoop(subCtx, userPubsub, "user")
+	go c.dispatchLoop(subCtx, broadcastPubsub, "broadcast")
+
+	return func() error {
+		subCancel()
+		if err := userPubsub.Close(); err != nil {
+			return err
+		}
+		return broadcastPubsub.Close()
+	}, nil
+}
+
+// dispatchLoop consumes Pub/Sub messages and dispatches them to the Hub.
+// channelType is "user" or "broadcast" for metric labeling.
+func (c *Client) dispatchLoop(ctx context.Context, ps *redis.PubSub, channelType string) {
+	ch := ps.Channel()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case msg, ok := <-ch:
+			if !ok {
+				return
+			}
+			start := time.Now()
+			c.handlePubSubMessage(msg, channelType)
+			c.metrics.ObservePubSubLatency("subscribe", time.Since(start).Seconds())
+		}
+	}
+}
+
+// handlePubSubMessage unmarshals the payload and dispatches to the Hub,
+// suppressing loopback (messages this instance published itself).
+func (c *Client) handlePubSubMessage(msg *redis.Message, channelType string) {
+	var crossMsg CrossInstanceMessage
+	if err := json.Unmarshal([]byte(msg.Payload), &crossMsg); err != nil {
+		observability.Logger().Warn("pubsub: invalid payload",
+			"channel", msg.Channel, "err", err)
+		return
+	}
+	// Loopback suppression: do not deliver our own publishes.
+	if crossMsg.Origin == c.instanceID {
+		return
+	}
+
+	payload, err := json.Marshal(map[string]any{
+		"type":      "message",
+		"event":     crossMsg.Event,
+		"data":      crossMsg.Data,
+		"timestamp": time.Now().UTC().Format(time.RFC3339),
+	})
+	if err != nil {
+		return
+	}
+
+	switch crossMsg.Kind {
+	case "user":
+		c.hub.SendToUser(crossMsg.UserID, payload)
+	case "broadcast":
+		c.hub.Broadcast(payload)
+	}
+}
+
+// ----- Idempotency (Kafka dedup) ------------------------------------------
+
+// DedupEventId performs a SETNX with TTL for Kafka event_id dedup.
+// Returns true when the event is fresh (first occurrence), false when it
+// has already been consumed.
+func (c *Client) DedupEventId(ctx context.Context, eventID string) (bool, error) {
+	if eventID == "" {
+		return true, nil
+	}
+	ok, err := c.rdb.SetNX(ctx, idempotentKey(eventID), "1", IdempotentTTL).Result()
+	if err != nil {
+		return false, err
+	}
+	return ok, nil
+}
+
+// ----- Key helpers ---------------------------------------------------------
+
+func onlineKey(userID string) string {
+	return "edu:push:online:" + userID
+}
+
+func userChannel(userID string) string {
+	return "edu:push:channel:user:" + userID
+}
+
+func broadcastChannel() string {
+	return "edu:push:channel:broadcast"
+}
+
+func idempotentKey(eventID string) string {
+	return "edu:push:idempotent:" + eventID
+}
+
+// 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 {
+	const prefix = "edu:push:channel:user:"
+	if !strings.HasPrefix(channel, prefix) {
+		return ""
+	}
+	return strings.TrimPrefix(channel, prefix)
+}