diff --git a/go.work b/go.work index bc8a5a6..2acfd4e 100644 --- a/go.work +++ b/go.work @@ -1,6 +1,7 @@ go 1.26.0 use ( + ./packages/shared-go ./services/api-gateway ./services/push-gateway ) diff --git a/go.work.sum b/go.work.sum index 271c4fc..6123523 100644 --- a/go.work.sum +++ b/go.work.sum @@ -1,35 +1,84 @@ cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.31.0/go.mod h1:P4WPRUkOhJC13W//jWpyfJNDAIpvRbAUIYLX/4jtlE0= +github.com/alecthomas/kingpin/v2 v2.4.0/go.mod h1:0gyi0zQnjuFk8xrkNKamJoyUo382HRL7ATRpFZCw6tE= +github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE= github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY= github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4= github.com/envoyproxy/go-control-plane v0.14.0/go.mod h1:NcS5X47pLl/hfqxU70yPwL9ZMkUlwlKxtAohpi2wBEU= github.com/envoyproxy/go-control-plane/envoy v1.37.0/go.mod h1:DReE9MMrmecPy+YvQOAOHNYMALuowAnbjjEMkkWOi6A= github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4= github.com/envoyproxy/protoc-gen-validate v1.3.3/go.mod h1:TsndJ/ngyIdQRhMcVVGDDHINPLWB7C82oDArY51KfB0= +github.com/go-chi/chi/v5 v5.0.12/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8= github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/golang/glog v1.2.5/go.mod h1:6AhwSGph0fcJtXVM/PEHPqZlFeoLxhs7/t5UDAwmO+w= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= github.com/jordanlewis/gcassert v0.0.0-20250430164644-389ef753e22e/go.mod h1:ZybsQk6DWyN5t7An1MuPm1gtSZ1xDaTXS9ZjIOxvQrk= +github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= +github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM= +github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM= -github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= -github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= -github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/segmentio/kafka-go v0.4.48/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg= github.com/spiffe/go-spiffe/v2 v2.6.0/go.mod h1:gm2SeUoMZEtpnzPNs2Csc0D/gX33k1xIx7lEzqblHEs= github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4= github.com/xdg-go/scram v1.2.0/go.mod h1:3dlrS0iBaWKYVt2ZfA4cj48umJZ+cAEbR6/SjLA88I8= github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= +github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU= github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.opentelemetry.io/contrib/detectors/gcp v1.42.0/go.mod h1:W9zQ439utxymRrXsUOzZbFX4JhLxXU4+ZnCt8GG7yA8= +go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= +go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= +go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= +golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo= +golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U= golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY= golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= +golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= diff --git a/packages/shared-go/go.mod b/packages/shared-go/go.mod index 8570fce..75e68b1 100644 --- a/packages/shared-go/go.mod +++ b/packages/shared-go/go.mod @@ -3,6 +3,24 @@ module github.com/edu-cloud/shared-go go 1.22 require ( + github.com/golang-jwt/jwt/v5 v5.2.1 + 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.opentelemetry.io/otel/trace v1.44.0 go.uber.org/zap v1.27.0 - github.com/go-chi/chi/v5 v5.0.12 +) + +require ( + github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect + golang.org/x/sys v0.45.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/grpc v1.81.1 // indirect + google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/services/push-gateway/go.mod b/services/push-gateway/go.mod index 34d3d71..ec9341a 100644 --- a/services/push-gateway/go.mod +++ b/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 diff --git a/services/push-gateway/internal/config/config.go b/services/push-gateway/internal/config/config.go index 7b835a6..e4e1b86 100644 --- a/services/push-gateway/internal/config/config.go +++ b/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 } diff --git a/services/push-gateway/internal/hub/hub.go b/services/push-gateway/internal/hub/hub.go index e4c44e5..f667a26 100644 --- a/services/push-gateway/internal/hub/hub.go +++ b/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: 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) + } } diff --git a/services/push-gateway/internal/kafkaconsumer/consumer.go b/services/push-gateway/internal/kafkaconsumer/consumer.go new file mode 100644 index 0000000..fdc835e --- /dev/null +++ b/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...); 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}) +} diff --git a/services/push-gateway/internal/observability/logger.go b/services/push-gateway/internal/observability/logger.go new file mode 100644 index 0000000..4d795cc --- /dev/null +++ b/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) +} diff --git a/services/push-gateway/internal/observability/metrics.go b/services/push-gateway/internal/observability/metrics.go new file mode 100644 index 0000000..e011ba8 --- /dev/null +++ b/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() +} diff --git a/services/push-gateway/internal/observability/tracer.go b/services/push-gateway/internal/observability/tracer.go index 3d1861c..9bb95c8 100644 --- a/services/push-gateway/internal/observability/tracer.go +++ b/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) } } } diff --git a/services/push-gateway/internal/redisclient/pubsub.go b/services/push-gateway/internal/redisclient/pubsub.go new file mode 100644 index 0000000..8e5b276 --- /dev/null +++ b/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: SET (members are instanceID) TTL 60s +// edu:push:session:: HASH (instanceID, device, lastPing) TTL 60s +// edu:push:channel:user: Pub/Sub channel (directed push) +// edu:push:channel:broadcast Pub/Sub channel (broadcast push) +// edu:push:idempotent: 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: 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: 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:. +// 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:" +// 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) +}