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

package grpcapi

import (
	"context"
	"errors"
	"testing"

	"galaxy/gateway/internal/session"
	gatewayv1 "galaxy/gateway/proto/galaxy/gateway/v1"

	"connectrpc.com/connect"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"google.golang.org/grpc"
)

func TestExecuteCommandRejectsUnknownSession(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{}
	server, runGateway := newTestGateway(t, ServerDependencies{
		Service: delegate,
		SessionCache: staticSessionCache{
			lookupFunc: func(context.Context, string) (session.Record, error) {
				return session.Record{}, session.ErrNotFound
			},
		},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	_, err := client.ExecuteCommand(context.Background(), connect.NewRequest(newValidExecuteCommandRequest()))
	require.Error(t, err)
	assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err))
	assert.Equal(t, "unknown device session", connectErrorMessage(t, err))
	assert.Zero(t, delegate.executeCalls)
}

func TestSubscribeEventsRejectsUnknownSession(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{}
	server, runGateway := newTestGateway(t, ServerDependencies{
		Service: delegate,
		SessionCache: staticSessionCache{
			lookupFunc: func(context.Context, string) (session.Record, error) {
				return session.Record{}, session.ErrNotFound
			},
		},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	err := subscribeEventsError(t, context.Background(), client, newValidSubscribeEventsRequest())
	require.Error(t, err)
	assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err))
	assert.Equal(t, "unknown device session", connectErrorMessage(t, err))
	assert.Zero(t, delegate.subscribeCalls)
}

func TestExecuteCommandRejectsRevokedSession(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{}
	server, runGateway := newTestGateway(t, ServerDependencies{
		Service:      delegate,
		SessionCache: staticSessionCache{lookupFunc: func(context.Context, string) (session.Record, error) { return newRevokedSessionRecord(), nil }},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	_, err := client.ExecuteCommand(context.Background(), connect.NewRequest(newValidExecuteCommandRequest()))
	require.Error(t, err)
	assert.Equal(t, connect.CodeFailedPrecondition, connect.CodeOf(err))
	assert.Equal(t, "device session is revoked", connectErrorMessage(t, err))
	assert.Zero(t, delegate.executeCalls)
}

func TestSubscribeEventsRejectsRevokedSession(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{}
	server, runGateway := newTestGateway(t, ServerDependencies{
		Service:      delegate,
		SessionCache: staticSessionCache{lookupFunc: func(context.Context, string) (session.Record, error) { return newRevokedSessionRecord(), nil }},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	err := subscribeEventsError(t, context.Background(), client, newValidSubscribeEventsRequest())
	require.Error(t, err)
	assert.Equal(t, connect.CodeFailedPrecondition, connect.CodeOf(err))
	assert.Equal(t, "device session is revoked", connectErrorMessage(t, err))
	assert.Zero(t, delegate.subscribeCalls)
}

func TestExecuteCommandRejectsSessionCacheUnavailable(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{}
	server, runGateway := newTestGateway(t, ServerDependencies{
		Service: delegate,
		SessionCache: staticSessionCache{
			lookupFunc: func(context.Context, string) (session.Record, error) {
				return session.Record{}, errors.New("redis down")
			},
		},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	_, err := client.ExecuteCommand(context.Background(), connect.NewRequest(newValidExecuteCommandRequest()))
	require.Error(t, err)
	assert.Equal(t, connect.CodeUnavailable, connect.CodeOf(err))
	assert.Equal(t, "session cache is unavailable", connectErrorMessage(t, err))
	assert.Zero(t, delegate.executeCalls)
}

func TestSubscribeEventsRejectsSessionCacheUnavailable(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{}
	server, runGateway := newTestGateway(t, ServerDependencies{
		Service: delegate,
		SessionCache: staticSessionCache{
			lookupFunc: func(context.Context, string) (session.Record, error) {
				return session.Record{}, errors.New("redis down")
			},
		},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	err := subscribeEventsError(t, context.Background(), client, newValidSubscribeEventsRequest())
	require.Error(t, err)
	assert.Equal(t, connect.CodeUnavailable, connect.CodeOf(err))
	assert.Equal(t, "session cache is unavailable", connectErrorMessage(t, err))
	assert.Zero(t, delegate.subscribeCalls)
}

func TestExecuteCommandAttachesResolvedSession(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{
		executeCommandFunc: func(ctx context.Context, req *gatewayv1.ExecuteCommandRequest) (*gatewayv1.ExecuteCommandResponse, error) {
			record, ok := resolvedSessionFromContext(ctx)
			require.True(t, ok)
			assert.Equal(t, newActiveSessionRecord(), record)
			return &gatewayv1.ExecuteCommandResponse{RequestId: req.GetRequestId()}, nil
		},
	}

	server, runGateway := newTestGateway(t, ServerDependencies{
		Service:      delegate,
		SessionCache: staticSessionCache{lookupFunc: func(context.Context, string) (session.Record, error) { return newActiveSessionRecord(), nil }},
		ReplayStore:  staticReplayStore{},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	response, err := client.ExecuteCommand(context.Background(), connect.NewRequest(newValidExecuteCommandRequest()))
	require.NoError(t, err)
	assert.Equal(t, "request-123", response.Msg.GetRequestId())
}

func TestSubscribeEventsAttachesResolvedSession(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{
		subscribeEventsFunc: func(req *gatewayv1.SubscribeEventsRequest, stream grpc.ServerStreamingServer[gatewayv1.GatewayEvent]) error {
			record, ok := resolvedSessionFromContext(stream.Context())
			require.True(t, ok)
			assert.Equal(t, newActiveSessionRecord(), record)
			return nil
		},
	}

	server, runGateway := newTestGateway(t, ServerDependencies{
		Service:      delegate,
		SessionCache: staticSessionCache{lookupFunc: func(context.Context, string) (session.Record, error) { return newActiveSessionRecord(), nil }},
		ReplayStore:  staticReplayStore{},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	stream, err := client.SubscribeEvents(context.Background(), connect.NewRequest(newValidSubscribeEventsRequest()))
	require.NoError(t, err)

	event := recvBootstrapEvent(t, stream)
	assertServerTimeBootstrapEvent(t, event, newTestResponseSignerPublicKey(), "request-123", "trace-123", testCurrentTime.UnixMilli())

	require.False(t, stream.Receive())
	require.NoError(t, stream.Err())
}

func TestSubscribeEventsAttachesAuthenticatedStreamBinding(t *testing.T) {
	t.Parallel()

	delegate := &recordingEdgeGatewayService{
		subscribeEventsFunc: func(req *gatewayv1.SubscribeEventsRequest, stream grpc.ServerStreamingServer[gatewayv1.GatewayEvent]) error {
			binding, ok := authenticatedStreamBindingFromContext(stream.Context())
			require.True(t, ok)
			assert.Equal(t, authenticatedStreamBinding{
				UserID:          "user-123",
				DeviceSessionID: "device-session-123",
				MessageType:     "gateway.subscribe",
				RequestID:       "request-123",
				TraceID:         "trace-123",
			}, binding)
			return nil
		},
	}

	server, runGateway := newTestGateway(t, ServerDependencies{
		Service:      delegate,
		SessionCache: staticSessionCache{lookupFunc: func(context.Context, string) (session.Record, error) { return newActiveSessionRecord(), nil }},
		ReplayStore:  staticReplayStore{},
	})
	defer runGateway.stop(t)

	addr := waitForListenAddr(t, server)
	client := newEdgeClient(t, addr)
	stream, err := client.SubscribeEvents(context.Background(), connect.NewRequest(newValidSubscribeEventsRequest()))
	require.NoError(t, err)

	event := recvBootstrapEvent(t, stream)
	assertServerTimeBootstrapEvent(t, event, newTestResponseSignerPublicKey(), "request-123", "trace-123", testCurrentTime.UnixMilli())

	require.False(t, stream.Receive())
	require.NoError(t, stream.Err())
}

type staticSessionCache struct {
	lookupFunc func(context.Context, string) (session.Record, error)
}

func (c staticSessionCache) Lookup(ctx context.Context, deviceSessionID string) (session.Record, error) {
	return c.lookupFunc(ctx, deviceSessionID)
}

func (staticSessionCache) MarkRevoked(string)           {}
func (staticSessionCache) MarkAllRevokedForUser(string) {}