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

// Package config loads runtime configuration for push-gateway from environment
// 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).
package config

import (
	"log"
	"os"
	"strings"

	"github.com/edu-cloud/shared-go/env"
)

// Config holds all push-gateway runtime configuration.
type Config struct {
	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 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 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 := 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)")
			jwtSecret = devJWTSecret
		} else {
			log.Fatal("JWT_SECRET must be set in non-dev mode")
		}
	}

	// PUSH_INTERNAL_TOKEN is the canonical env var for /internal/* auth
	// (ARB-013 alignment with msg). INTERNAL_API_TOKEN is kept as a
	// backward-compat alias for existing deployments.
	internalToken := env.Get("PUSH_INTERNAL_TOKEN", "")
	if internalToken == "" {
		internalToken = env.Get("INTERNAL_API_TOKEN", "")
	}
	if internalToken == "" && !devMode {
		log.Fatal("PUSH_INTERNAL_TOKEN (or INTERNAL_API_TOKEN) must be set in non-dev mode")
	}

	return &Config{
		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")),
		// 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()),
	}
}

// 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
	}
	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
}