Files
Datarush/internal/lms/server/server.go
T

183 lines
4.6 KiB
Go

package server
import (
"context"
"fmt"
"log"
"net"
"net/http"
"time"
"datarush/internal/lms/config"
"datarush/internal/lms/interceptor"
httpHandlers "datarush/internal/lms/handler/http"
authPb "datarush/pkg/api/auth"
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq" // postgres driver
"github.com/redis/go-redis/v9"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/reflection"
)
const (
httpReadTimeout = 10 * time.Second
httpWriteTimeout = 10 * time.Second
httpIdleTimeout = 60 * time.Second
redisRetryCount = 2
redisMinRetryBackoff = 50 * time.Millisecond
redisMaxRetryBackoff = 200 * time.Millisecond
redisDialTimeout = 1 * time.Second
redisDialerRetries = 3
redisTimeout = 2 * time.Second
)
type Server struct {
grpcServer *grpc.Server
config *config.Config
db *sqlx.DB
redisDB *redis.Client
}
func New(cfg *config.Config) *Server {
loggerInterceptor := interceptor.NewLoggerInterceptor()
grpcServer := grpc.NewServer(
grpc.UnaryInterceptor(loggerInterceptor.Unary()),
grpc.StreamInterceptor(loggerInterceptor.Stream()),
)
return &Server{
grpcServer: grpcServer,
config: cfg,
}
}
func runHTTPHandler(s *Server, grpcServerEndpoint *string) error {
ctx := context.Background()
ctx, cancel := context.WithCancel(ctx)
defer cancel()
gwmux := runtime.NewServeMux()
opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
// Register auth service
if err := registerAuthService(ctx, gwmux, s.config.AuthGRPCAddr, opts); err != nil {
log.Printf("failed to register auth service: %v", err)
}
mux := http.NewServeMux()
mux.Handle("/healthz", httpHandlers.NewHealthHandler(s.db, s.redisDB))
mux.Handle("/", gwmux)
srv := &http.Server{
Addr: fmt.Sprintf(":%d", s.config.HTTPPort),
Handler: mux,
ReadTimeout: httpReadTimeout,
WriteTimeout: httpWriteTimeout,
IdleTimeout: httpIdleTimeout,
}
return srv.ListenAndServe()
}
func registerAuthService(ctx context.Context, gwmux *runtime.ServeMux, authAddr string, opts []grpc.DialOption) error {
if err := authPb.RegisterAuthServiceHandlerFromEndpoint(ctx, gwmux, authAddr, opts); err != nil {
return fmt.Errorf("register auth service handler: %w", err)
}
log.Printf("registered auth service from %s", authAddr)
return nil
}
func registerAuthHandlerFromEndpoint(ctx context.Context, gwmux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) error {
// We'll import the proto and register it here
// For now, this is a placeholder that will be called from gateway integration
return nil
}
func getDatabase(cfg config.Config) (*sqlx.DB, error) {
db, err := sqlx.Connect("postgres", cfg.BuildPostgresConnStr())
if err != nil {
return nil, fmt.Errorf("connect to database: %w", err)
}
return db, nil
}
func getRedis(cfg config.Config) (*redis.Client, error) {
conn, err := redis.ParseURL(cfg.RedisURI)
client := redis.NewClient(&redis.Options{
Addr: conn.Addr,
MaxRetries: redisRetryCount,
MinRetryBackoff: redisMinRetryBackoff,
MaxRetryBackoff: redisMaxRetryBackoff,
DialTimeout: redisDialTimeout,
DialerRetries: redisDialerRetries,
DialerRetryTimeout: redisDialTimeout,
ReadTimeout: redisTimeout,
WriteTimeout: redisTimeout,
})
if err != nil {
return nil, fmt.Errorf("parse Redis URI: %w", err)
}
_, err = client.Ping(context.Background()).Result()
if err != nil {
return nil, fmt.Errorf("connect to Redis server: %w", err)
}
return client, nil
}
func (s *Server) RegisterServices() {
db, err := getDatabase(*s.config)
if err != nil {
log.Print(err)
}
s.db = db
redisDB, err := getRedis(*s.config)
if err != nil {
log.Print(err)
}
s.redisDB = redisDB
if s.config.GRPCEnableReflection {
reflection.Register(s.grpcServer)
log.Println("gRPC server will start with reflection")
}
}
func (s *Server) Start() error {
addr := fmt.Sprintf(":%d", s.config.GRPCPort)
lis, err := net.Listen("tcp", addr)
if err != nil {
return fmt.Errorf("failed to listen: %w", err)
}
if s.config.EnableHTTPHandler {
go func() {
log.Printf("starting HTTP gateway on port %d", s.config.HTTPPort)
if err := runHTTPHandler(s, &addr); err != nil {
log.Printf("HTTP gateway failed: %v", err)
}
}()
}
log.Printf("starting gRPC server on port %d", s.config.GRPCPort)
if err := s.grpcServer.Serve(lis); err != nil {
return fmt.Errorf("failed to serve: %w", err)
}
return nil
}
func (s *Server) Stop() {
s.grpcServer.GracefulStop()
log.Println("gRPC server stopped gracefully")
}