Edu/services/push-gateway/main.go

// Command push-gateway is the Edu platform's real-time push gateway.
//
// M7 (ADR-040): The gateway no longer mounts Kafka directly. Real-time push
// is routed through a Redis Pub/Sub backplane. The msg service publishes
// notifications to Redis channel `user:{userId}:notify` after Kafka
// consumption (persistence, desensitization). This gateway dynamically
// subscribes to the user's Redis channel when an SSE connection is opened,
// and unsubscribes on disconnect.
//
// Endpoints:
//   - GET /sse/notifications: SSE stream for real-time notifications (JWT auth,
//     per-connection Redis subscription). M7 preferred path.
//   - GET /ws: WebSocket upgrade (JWT auth, legacy path with global Redis
//     PSubscribe fanout for backwards compatibility).
//   - POST /internal/push, /internal/broadcast: internal HTTP APIs (X-Internal-Key
//     auth, consumed by msg for legacy WebSocket fanout).
//   - GET /internal/online/:userID: presence query.
//   - GET /healthz, /readyz: health probes.
//   - GET /metrics: Prometheus metrics.
//
// Startup sequence:
//  1. config.Load (env vars, DevMode detection)
//  2. observability.InitLogger (slog JSON/text)
//  3. observability.InitTracer (shared-go/tracer → OTLP)
//  4. observability.NewMetrics (Prometheus)
//  5. hub.NewHub (in-memory connection registry for WebSocket)
//  6. redisclient.New + SetHub + RebuildPresenceOnStartup (ISSUE-058)
//  7. redisclient.SubscribeAll (cross-instance Pub/Sub fanout for WebSocket)
//  8. ws.NewHandler (JWT RS256, X-Internal-Key, WebSocket upgrade)
//  9. sse.NewHandler (JWT RS256, per-connection Redis subscription)
//  10. health.NewReadyzer (/readyz soft-failure probe)
//  11. gin router + http.Server
//
// Graceful shutdown (SIGINT/SIGTERM):
//  1. Hub.CloseAll (send close frame 1001 to every live WebSocket connection)
//  2. http.Server.Shutdown (stop accepting new HTTP/SSE/WebSocket requests)
//  3. Redis Pub/Sub cancel + client.Close
//  4. Tracer shutdown (flush pending spans)
package main

import (
	"context"
	"errors"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/edu-cloud/push-gateway/internal/config"
	"github.com/edu-cloud/push-gateway/internal/health"
	"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/edu-cloud/push-gateway/internal/sse"
	"github.com/edu-cloud/push-gateway/internal/ws"
	"github.com/gin-gonic/gin"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
)

const serviceName = "push-gateway"

// shutdownTimeout bounds the graceful shutdown of the HTTP server. After this
// window in-flight requests are forcibly closed. WebSocket long-poll
// connections and SSE streams are drained by Hub.CloseAll before this timer
// starts; SSE streams are drained by http.Server.Shutdown.
const shutdownTimeout = 10 * time.Second

func main() {
	cfg := config.Load()

	// 1. Structured logger (slog).
	logger := observability.InitLogger(cfg.DevMode)
	logger.Info("push-gateway starting",
		"dev_mode", cfg.DevMode,
		"instance_id", cfg.InstanceID,
		"port", cfg.Port,
		"max_conns_per_user", cfg.MaxConnsPerUser,
		"jwks_url", cfg.JWKSURL,
		"redis_url", cfg.RedisURL,
	)

	// 2. OpenTelemetry tracer (best-effort).
	tracerShutdown := observability.InitTracer(serviceName, cfg.OTLPEndpoint)
	defer tracerShutdown()

	// 3. Prometheus metrics.
	metrics := observability.NewMetrics()

	// 4. Hub: in-memory connection registry (for WebSocket legacy path).
	h := hub.NewHub(cfg.MaxConnsPerUser)

	// 5. Redis client (optional in DevMode). When present, wire presence
	// hooks to the Hub and rebuild the online SET (ISSUE-058).
	var redisClient *redisclient.Client
	if cfg.RedisURL != "" {
		rc, err := redisclient.New(cfg.RedisURL, cfg.InstanceID, metrics)
		if err != nil {
			logger.Error("redis connect failed; running without cross-instance fanout",
				"err", err, "url", cfg.RedisURL)
		} else {
			redisClient = rc
			redisClient.SetHub(h)
			rebuildCtx, rebuildCancel := context.WithTimeout(context.Background(), 5*time.Second)
			if err := redisClient.RebuildPresenceOnStartup(rebuildCtx); err != nil {
				logger.Warn("redis presence rebuild failed (continuing)", "err", err)
			}
			rebuildCancel()
			logger.Info("redis presence synced", "url", cfg.RedisURL)
		}
	} else {
		logger.Warn("REDIS_URL empty; SSE and cross-instance fanout disabled")
	}

	// 6. Redis Pub/Sub subscriber (cross-instance message fanout for WebSocket).
	var pubsubCancel func() error
	if redisClient != nil {
		subCtx, subCancel := context.WithCancel(context.Background())
		defer subCancel()
		var err error
		pubsubCancel, err = redisClient.SubscribeAll(subCtx)
		if err != nil {
			logger.Error("redis SubscribeAll failed; cross-instance fanout disabled",
				"err", err)
		} else {
			logger.Info("redis pubsub subscriber active")
		}
	}

	// 7. WebSocket + internal HTTP handlers (legacy path, backwards compat).
	wsHandler := ws.NewHandler(h, cfg, redisClient, metrics)

	// 8. SSE handler (M7 preferred path: per-connection Redis subscription).
	sseHandler := sse.NewHandler(cfg, redisClient, metrics)

	// 9. /readyz probe (soft failure per ARB-015 §17.4).
	readyzer := health.NewReadyzer(h, redisClient, serviceName, cfg.InstanceID)

	// 10. Gin router.
	gin.SetMode(gin.ReleaseMode)
	r := gin.New()
	r.Use(gin.Recovery())
	r.Use(otelgin.Middleware(serviceName))

	// Liveness (no auth, no dependency checks).
	r.GET("/healthz", health.Healthz(serviceName))
	// Readiness (soft failure on Redis).
	r.GET("/readyz", readyzer.Handler)
	// Prometheus metrics.
	r.GET("/metrics", gin.WrapH(promhttp.Handler()))
	// SSE endpoint (M7 preferred: per-connection Redis subscription).
	r.GET("/sse/notifications", sseHandler.HandleNotifications)
	// WebSocket upgrade (JWT auth via query ?token= or Authorization header).
	r.GET("/ws", wsHandler.HandleWebSocket)
	// Internal HTTP APIs consumed by msg (X-Internal-Key auth, ARB-013).
	internal := r.Group("/internal")
	internal.POST("/push", wsHandler.PushHandler)
	internal.POST("/broadcast", wsHandler.BroadcastHandler)
	internal.GET("/online/:userID", wsHandler.OnlineHandler)

	srv := &http.Server{
		Addr:         ":" + cfg.Port,
		Handler:      r,
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 0, // SSE and WebSocket connections are long-lived; no write timeout.
	}

	// 11. Start HTTP server.
	go func() {
		logger.Info("http server listening", "addr", srv.Addr)
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			logger.Error("http server fatal error", "err", err)
			os.Exit(1)
		}
	}()

	// 12. Wait for termination signal.
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	sig := <-quit
	logger.Info("shutdown signal received", "signal", sig.String())

	// 13. Graceful shutdown sequence (see package doc).
	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), shutdownTimeout)
	defer shutdownCancel()

	// 13a. Hub.CloseAll sends close frame 1001 to every live WebSocket connection.
	h.CloseAll()
	logger.Info("hub closeAll complete; draining connections",
		"active_connections", h.ActiveConnections())

	// 13b. Stop accepting new HTTP requests; drain in-flight (including SSE).
	if err := srv.Shutdown(shutdownCtx); err != nil {
		logger.Error("http server shutdown error", "err", err)
	}

	// 13c. Stop Redis Pub/Sub subscriber + close client.
	if pubsubCancel != nil {
		if err := pubsubCancel(); err != nil {
			logger.Warn("redis pubsub cancel error", "err", err)
		}
	}
	if redisClient != nil {
		if err := redisClient.Close(); err != nil {
			logger.Warn("redis close error", "err", err)
		}
	}

	logger.Info("push-gateway shutdown complete")
}