grpc_router.go

/*
 * Copyright (c) 2024.
 * Lizenziert unter der EUPL, Version 1.2 oder - sobald
 * diese von der Europäischen Kommission genehmigt wurden -
 * Folgeversionen der EUPL ("Lizenz");
 * Sie dürfen dieses Werk ausschließlich gemäß
 * dieser Lizenz nutzen.
 * Eine Kopie der Lizenz finden Sie hier:
 *
 * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
 *
 * Sofern nicht durch anwendbare Rechtsvorschriften
 * gefordert oder in schriftlicher Form vereinbart, wird
 * die unter der Lizenz verbreitete Software "so wie sie
 * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
 * ausdrücklich oder stillschweigend - verbreitet.
 * Die sprachspezifischen Genehmigungen und Beschränkungen
 * unter der Lizenz sind dem Lizenztext zu entnehmen.
 */

package server

import (
	pb "antragsraum-proxy/gen/go"
	"antragsraum-proxy/internal/config"
	"context"
	"errors"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	log "github.com/sirupsen/logrus"
	"net"
)

type rueckfrageRouter struct {
	pb.UnimplementedAntragraumServiceServer
}

func (s *rueckfrageRouter) FindRueckfragen(ctx context.Context, in *pb.GrpcFindRueckfragenRequest) (*pb.GrpcFindRueckfragenResponse, error) {
	response, err := s.handleRequest(ctx, func(client pb.AntragraumServiceClient, ctx context.Context) (interface{}, error) {
		return client.FindRueckfragen(ctx, in)
	})

	if response == nil {
		return nil, err
	}

	return response.(*pb.GrpcFindRueckfragenResponse), err
}

func (s *rueckfrageRouter) GetRueckfrage(ctx context.Context, in *pb.GrpcGetRueckfrageRequest) (*pb.GrpcGetRueckfrageResponse, error) {
	response, err := s.handleRequest(ctx, func(client pb.AntragraumServiceClient, ctx context.Context) (interface{}, error) {
		return client.GetRueckfrage(ctx, in)
	})

	if response == nil {
		return nil, err
	}

	return response.(*pb.GrpcGetRueckfrageResponse), err
}

func (s *rueckfrageRouter) SendRueckfrageAnswer(ctx context.Context, in *pb.GrpcSendRueckfrageAnswerRequest) (*pb.GrpcSendRueckfrageAnswerResponse, error) {
	response, err := s.handleRequest(ctx, func(client pb.AntragraumServiceClient, ctx context.Context) (interface{}, error) {
		return client.SendRueckfrageAnswer(ctx, in)
	})

	if response == nil {
		return nil, err
	}

	return response.(*pb.GrpcSendRueckfrageAnswerResponse), err
}

func (s *rueckfrageRouter) GetAttachmentMetadata(ctx context.Context, in *pb.GrpcGetAttachmentMetadataRequest) (*pb.GrpcGetAttachmentMetadataResponse, error) {
	response, err := s.handleRequest(ctx, func(client pb.AntragraumServiceClient, ctx context.Context) (interface{}, error) {
		return client.GetAttachmentMetadata(ctx, in)
	})

	if response == nil {
		return nil, err
	}

	return response.(*pb.GrpcGetAttachmentMetadataResponse), err
}

func (s *rueckfrageRouter) handleRequest(ctx context.Context, handler func(pb.AntragraumServiceClient, context.Context) (interface{}, error)) (interface{}, error) {
	return handleRequest(ctx, createRueckfrageClient, handler)
}

func createRueckfrageClient(grpcAddress string) (pb.AntragraumServiceClient, func() error, error) {
	target := GetGrpcServerUrl(grpcAddress, conf)
	conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.WithError(err).Error(fmt.Sprintf("rueckfrage router failed to route: %v", err))

		return nil, nil, errors.New("rueckfrage router failed to route")
	}

	return pb.NewAntragraumServiceClient(conn), conn.Close, nil
}

type commandRouter struct {
	pb.UnimplementedCommandServiceServer
}

func (s *commandRouter) GetCommand(ctx context.Context, in *pb.GrpcGetCommandRequest) (*pb.GrpcCommand, error) {
	response, err := s.handleRequest(ctx, func(client pb.CommandServiceClient, ctx context.Context) (interface{}, error) {
		return client.GetCommand(ctx, in)
	})

	if response == nil {
		return nil, err
	}

	return response.(*pb.GrpcCommand), err
}

func (s *commandRouter) handleRequest(ctx context.Context, handler func(pb.CommandServiceClient, context.Context) (interface{}, error)) (interface{}, error) {
	return handleRequest(ctx, createCommandClient, handler)
}

func createCommandClient(grpcAddress string) (pb.CommandServiceClient, func() error, error) {
	target := GetGrpcServerUrl(grpcAddress, conf)
	conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Error(fmt.Sprintf("command router failed to route: %v", err))

		return nil, nil, errors.New("command router failed to route")
	}

	return pb.NewCommandServiceClient(conn), conn.Close, nil
}

func GetGrpcServerUrl(grpcAddress string, c config.Config) string {
	return fmt.Sprintf("%v:%d", grpcAddress, c.Grpc.Server.Port)
}

func handleRequest[T any](
	ctx context.Context,
	createClientFunc func(grpcAddress string) (T, func() error, error),
	handler func(T, context.Context) (interface{}, error),
) (interface{}, error) {
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return nil, status.Error(codes.InvalidArgument, "unable to retrieve metadata")
	}

	grpcAddress := md.Get(GrpcAddressMetadata)
	if len(grpcAddress) == 0 {
		return nil, status.Error(codes.InvalidArgument, "grpc address is missing")
	}

	client, connCleanUp, err := createClientFunc(grpcAddress[0])
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	defer connCleanUp()

	return handler(client, ctx)
}

func StartGrpcRouter() *grpc.Server {
	s := grpc.NewServer()
	pb.RegisterAntragraumServiceServer(s, &rueckfrageRouter{})
	pb.RegisterCommandServiceServer(s, &commandRouter{})

	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", conf.Grpc.Router.Port))
	if err != nil {
		log.Fatal(fmt.Sprintf("gRPC router failed to listen: %v", err))
	}
	log.Info(fmt.Sprintf("gRPC router listening on port %d", conf.Grpc.Router.Port))

	if err = s.Serve(lis); err != nil {
		log.Fatal(fmt.Sprintf("gRPC router failed to serve: %v", err))
	}

	return s
}