galaxy-game/gateway/internal/grpcapi/command_routing.go

package grpcapi

import (
	"bytes"
	"context"
	"crypto/sha256"
	"errors"
	"strings"
	"time"

	"galaxy/gateway/authn"
	"galaxy/gateway/internal/clock"
	"galaxy/gateway/internal/downstream"
	edgev1 "galaxy/gateway/proto/edge/v1"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// commandRoutingService translates the verified authenticated request context
// into an internal downstream command and signs successful unary responses.
type commandRoutingService struct {
	edgev1.UnimplementedGatewayServer

	subscribeDelegate edgev1.GatewayServer
	router            downstream.Router
	responseSigner    authn.ResponseSigner
	clock             clock.Clock
	downstreamTimeout time.Duration
}

// ExecuteCommand builds a verified downstream command, routes it by exact
// message_type, executes it, and signs the resulting unary response.
func (s commandRoutingService) ExecuteCommand(ctx context.Context, _ *edgev1.ExecuteCommandRequest) (*edgev1.ExecuteCommandResponse, error) {
	command, err := authenticatedCommandFromContext(ctx)
	if err != nil {
		return nil, err
	}

	client, err := s.router.Route(command.MessageType)
	switch {
	case err == nil:
	case errors.Is(err, downstream.ErrRouteNotFound):
		return nil, status.Error(codes.Unimplemented, "message_type is not routed")
	case errors.Is(err, downstream.ErrDownstreamUnavailable):
		return nil, status.Error(codes.Unavailable, "downstream service is unavailable")
	default:
		return nil, status.Error(codes.Internal, "downstream route resolution failed")
	}

	downstreamCtx, cancel := context.WithTimeout(ctx, s.downstreamTimeout)
	defer cancel()

	result, err := client.ExecuteCommand(downstreamCtx, command)
	switch {
	case err == nil:
	case errors.Is(err, downstream.ErrDownstreamUnavailable),
		errors.Is(err, context.DeadlineExceeded),
		errors.Is(err, context.Canceled):
		return nil, status.Error(codes.Unavailable, "downstream service is unavailable")
	default:
		return nil, status.Error(codes.Internal, "downstream execution failed")
	}

	if strings.TrimSpace(result.ResultCode) == "" {
		return nil, status.Error(codes.Internal, "downstream response is invalid")
	}

	responseTimestampMS := s.clock.Now().UTC().UnixMilli()
	payloadHash := sha256.Sum256(result.PayloadBytes)
	signature, err := s.responseSigner.SignResponse(authn.ResponseSigningFields{
		ProtocolVersion: command.ProtocolVersion,
		RequestID:       command.RequestID,
		TimestampMS:     responseTimestampMS,
		ResultCode:      result.ResultCode,
		PayloadHash:     payloadHash[:],
	})
	if err != nil {
		return nil, status.Error(codes.Unavailable, "response signer is unavailable")
	}

	return &edgev1.ExecuteCommandResponse{
		ProtocolVersion: command.ProtocolVersion,
		RequestId:       command.RequestID,
		TimestampMs:     responseTimestampMS,
		ResultCode:      result.ResultCode,
		PayloadBytes:    bytes.Clone(result.PayloadBytes),
		PayloadHash:     bytes.Clone(payloadHash[:]),
		Signature:       signature,
	}, nil
}

// SubscribeEvents delegates to the authenticated streaming service
// implementation selected during server construction.
func (s commandRoutingService) SubscribeEvents(req *edgev1.SubscribeEventsRequest, stream grpc.ServerStreamingServer[edgev1.GatewayEvent]) error {
	return s.subscribeDelegate.SubscribeEvents(req, stream)
}

// newCommandRoutingService constructs the final authenticated service that
// owns verified unary routing while preserving the delegated streaming path.
func newCommandRoutingService(subscribeDelegate edgev1.GatewayServer, router downstream.Router, responseSigner authn.ResponseSigner, clk clock.Clock, downstreamTimeout time.Duration) edgev1.GatewayServer {
	return commandRoutingService{
		subscribeDelegate: subscribeDelegate,
		router:            router,
		responseSigner:    responseSigner,
		clock:             clk,
		downstreamTimeout: downstreamTimeout,
	}
}

func authenticatedCommandFromContext(ctx context.Context) (downstream.AuthenticatedCommand, error) {
	envelope, ok := parsedEnvelopeFromContext(ctx)
	if !ok {
		return downstream.AuthenticatedCommand{}, status.Error(codes.Internal, "authenticated request context is incomplete")
	}

	record, ok := resolvedSessionFromContext(ctx)
	if !ok {
		return downstream.AuthenticatedCommand{}, status.Error(codes.Internal, "authenticated request context is incomplete")
	}

	return downstream.AuthenticatedCommand{
		ProtocolVersion: envelope.ProtocolVersion,
		UserID:          record.UserID,
		DeviceSessionID: record.DeviceSessionID,
		MessageType:     envelope.MessageType,
		TimestampMS:     envelope.TimestampMS,
		RequestID:       envelope.RequestID,
		TraceID:         envelope.TraceID,
		PayloadBytes:    bytes.Clone(envelope.PayloadBytes),
	}, nil
}

type unavailableResponseSigner struct{}

func (unavailableResponseSigner) SignResponse(authn.ResponseSigningFields) ([]byte, error) {
	return nil, errors.New("response signer is unavailable")
}

func (unavailableResponseSigner) SignEvent(authn.EventSigningFields) ([]byte, error) {
	return nil, errors.New("response signer is unavailable")
}

var _ edgev1.GatewayServer = commandRoutingService{}