[QoS][Cosmos] add minimal websockets request context to allow WS on Cosmos (#375)

## 🌿 Summary

Add minimal WebSocket request context validation to enable WebSocket
connections for Cosmos services.

### 🌱 Primary Changes:
- Implement `validateWebsocketRequest` to handle WebSocket upgrade
requests and RPC type validation
- Add specialized `buildWebsocketRequestContext` to create
WebSocket-specific request contexts
- Introduce error handling for unsupported WebSocket RPC types via
`createWebsocketUnsupportedRPCTypeContext`

### 🍃 Secondary changes:
- Add new file `request_validator_websocket.go` to encapsulate WebSocket
validation logic
- Include request observation building for WebSocket connections
- Update request context structure to accommodate WebSocket-specific
fields

## 🛠️ Type of change

Select one or more from the following:

- [x] New feature, functionality or library
- [x] Bug fix
- [ ] Code health or cleanup
- [ ] Documentation
- [ ] Other (specify)

---------

Co-authored-by: Arash Deshmeh <[email protected]>
Co-authored-by: Daniel Olshansky <[email protected]>

Author: 3 people

qos/cosmos/qos.go

@@ -90,10 +90,7 @@ func (qos *QoS) ParseHTTPRequest(_ context.Context, req *http.Request) (gateway.
 //
 // Implements gateway.QoSService interface.
 func (qos *QoS) ParseWebsocketRequest(_ context.Context) (gateway.RequestQoSContext, bool) {
-	return &requestContext{
-		logger:       qos.logger,
-		serviceState: qos.serviceState,
-	}, true
+	return qos.validateWebsocketRequest()
 }
 
 // HydrateDisqualifiedEndpointsResponse hydrates the disqualified endpoint response with the QoS-specific data.

qos/cosmos/request_validator_jsonrpc.go

@@ -3,6 +3,7 @@ package cosmos
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 
 	"github.com/pokt-network/poktroll/pkg/polylog"
@@ -48,7 +49,7 @@ func (rv *requestValidator) validateJSONRPCRequest(
 
 	// Hydrate the logger with data extracted from the request.
 	logger = logger.With(
-		"detected_rpc_type", rpcType.String(),
+		"rpc_type", rpcType.String(),
 		"jsonrpc_method", method,
 	)
 
@@ -281,7 +282,7 @@ func (rv *requestValidator) createJSONRPCUnsupportedRPCTypeObservation(
 		},
 		RequestLevelError: &qosobservations.RequestError{
 			ErrorKind:      qosobservations.RequestErrorKind_REQUEST_ERROR_USER_ERROR_JSONRPC_UNSUPPORTED_RPC_TYPE,
-			ErrorDetails:   "Unsupported RPC type: " + rpcType.String(),
+			ErrorDetails:   fmt.Sprintf("Unsupported RPC type %s for service %s", rpcType.String(), string(rv.serviceID)),
 			HttpStatusCode: int32(jsonrpcResponse.GetRecommendedHTTPStatusCode()),
 		},
 	}

qos/cosmos/request_validator_rest.go

@@ -36,7 +36,7 @@ func (rv *requestValidator) validateRESTRequest(
 	rpcType := determineRESTRPCType(httpRequestPath)
 
 	logger = logger.With(
-		"detected_rpc_type", rpcType.String(),
+		"rpc_type", rpcType.String(),
 		"request_path", httpRequestPath,
 	)
 

qos/cosmos/request_validator_websocket.go

@@ -0,0 +1,122 @@
+package cosmos
+
+import (
+	"errors"
+	"fmt"
+
+	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
+
+	"github.com/buildwithgrove/path/gateway"
+	qosobservations "github.com/buildwithgrove/path/observation/qos"
+	"github.com/buildwithgrove/path/qos"
+	"github.com/buildwithgrove/path/qos/jsonrpc"
+)
+
+// TODO_IMPROVE(@commoddity): Add endpoint-level QoS checks to determine WebSocket support.
+// Currently validates WebSocket upgrade requests at the service level only.
+
+// validateWebsocketRequest validates WebSocket upgrade requests for Cosmos SDK services.
+// Returns (requestContext, true) if WebSocket is supported.
+// Returns (errorContext, false) if WebSocket is not configured for this service.
+func (rv *requestValidator) validateWebsocketRequest() (gateway.RequestQoSContext, bool) {
+	logger := rv.logger.With(
+		"validator", "WebSocket",
+		"method", "validateWebsocketRequest",
+	)
+	rpcType := sharedtypes.RPCType_WEBSOCKET
+	logger = logger.With("rpc_type", rpcType.String())
+
+	// Verify WebSocket support in service configuration
+	if _, supported := rv.supportedAPIs[sharedtypes.RPCType_WEBSOCKET]; !supported {
+		logger.Warn().Msg("Request uses unsupported WebSocket RPC type")
+		return rv.createWebsocketUnsupportedRPCTypeContext(rpcType), false
+	}
+
+	// Build and return the request context
+	return rv.buildWebsocketRequestContext(
+		rpcType,
+		qosobservations.RequestOrigin_REQUEST_ORIGIN_ORGANIC,
+	), true
+}
+
+// buildWebsocketRequestContext builds a request context for WebSocket upgrade requests.
+func (rv *requestValidator) buildWebsocketRequestContext(
+	rpcType sharedtypes.RPCType,
+	requestOrigin qosobservations.RequestOrigin,
+) gateway.RequestQoSContext {
+	logger := rv.logger.With(
+		"method", "buildWebsocketRequestContext",
+	)
+	requestObservation := rv.buildWebsocketRequestObservations(
+		rpcType,
+		requestOrigin,
+	)
+	return &requestContext{
+		logger:                          logger,
+		serviceState:                    rv.serviceState,
+		observations:                    requestObservation,
+		protocolErrorObservationBuilder: buildProtocolErrorObservation,
+	}
+}
+
+// buildWebsocketRequestObservations builds a request observation for WebSocket upgrade requests.
+func (rv *requestValidator) buildWebsocketRequestObservations(
+	rpcType sharedtypes.RPCType,
+	requestOrigin qosobservations.RequestOrigin,
+) *qosobservations.CosmosRequestObservations {
+
+	return &qosobservations.CosmosRequestObservations{
+		CosmosChainId: rv.cosmosChainID,
+		ServiceId:     string(rv.serviceID),
+		RequestOrigin: requestOrigin,
+		RequestProfile: &qosobservations.CosmosRequestProfile{
+			BackendServiceDetails: &qosobservations.BackendServiceDetails{
+				BackendServiceType: convertToProtoBackendServiceType(rpcType),
+				SelectionReason:    "WebSocket upgrade request detection",
+			},
+		},
+	}
+}
+
+// createWebsocketUnsupportedRPCTypeContext creates error context when WebSocket is not configured
+func (rv *requestValidator) createWebsocketUnsupportedRPCTypeContext(
+	rpcType sharedtypes.RPCType,
+) gateway.RequestQoSContext {
+	err := errors.New("WebSocket not supported for this service")
+	response := jsonrpc.NewErrResponseInvalidRequest(jsonrpc.ID{}, err)
+
+	observations := rv.createWebsocketUnsupportedRPCTypeObservation(rpcType, response)
+
+	return &qos.RequestErrorContext{
+		Logger:   rv.logger,
+		Response: response,
+		Observations: &qosobservations.Observations{
+			ServiceObservations: &qosobservations.Observations_Cosmos{
+				Cosmos: observations,
+			},
+		},
+	}
+}
+
+// createWebsocketUnsupportedRPCTypeObservation creates an observation for unsupported WebSocket requests
+func (rv *requestValidator) createWebsocketUnsupportedRPCTypeObservation(
+	rpcType sharedtypes.RPCType,
+	jsonrpcResponse jsonrpc.Response,
+) *qosobservations.CosmosRequestObservations {
+	return &qosobservations.CosmosRequestObservations{
+		ServiceId:     string(rv.serviceID),
+		CosmosChainId: rv.cosmosChainID,
+		RequestOrigin: qosobservations.RequestOrigin_REQUEST_ORIGIN_ORGANIC,
+		RequestProfile: &qosobservations.CosmosRequestProfile{
+			BackendServiceDetails: &qosobservations.BackendServiceDetails{
+				BackendServiceType: convertToProtoBackendServiceType(rpcType),
+				SelectionReason:    "WebSocket upgrade request detection (unsupported)",
+			},
+		},
+		RequestLevelError: &qosobservations.RequestError{
+			ErrorKind:      qosobservations.RequestErrorKind_REQUEST_ERROR_USER_ERROR_JSONRPC_UNSUPPORTED_RPC_TYPE,
+			ErrorDetails:   fmt.Sprintf("Unsupported RPC type %s for service %s", rpcType.String(), string(rv.serviceID)),
+			HttpStatusCode: int32(jsonrpcResponse.GetRecommendedHTTPStatusCode()),
+		},
+	}
+}

qos/cosmos/request_validator_websocket_test.go

@@ -0,0 +1,68 @@
+package cosmos
+
+import (
+	"testing"
+
+	"github.com/pokt-network/poktroll/pkg/polylog/polyzero"
+	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_validateWebsocketRequest(t *testing.T) {
+	tests := []struct {
+		name                   string
+		supportedAPIs          map[sharedtypes.RPCType]struct{}
+		expectSuccess          bool
+		expectErrorContextType bool
+	}{
+		{
+			name: "supported websockets service config",
+			supportedAPIs: map[sharedtypes.RPCType]struct{}{
+				sharedtypes.RPCType_WEBSOCKET: {},
+			},
+			expectSuccess:          true,
+			expectErrorContextType: false,
+		},
+		{
+			name: "unsupported websockets service config",
+			supportedAPIs: map[sharedtypes.RPCType]struct{}{
+				sharedtypes.RPCType_REST: {},
+			},
+			expectSuccess:          false,
+			expectErrorContextType: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Set up the request validator with test data
+			validator := &requestValidator{
+				logger:        polyzero.NewLogger(),
+				cosmosChainID: "test-chain",
+				serviceID:     "test-service",
+				supportedAPIs: tt.supportedAPIs,
+				serviceState:  &serviceState{}, // minimal setup for context building
+			}
+
+			// Call the function under test
+			ctx, success := validator.validateWebsocketRequest()
+
+			// Verify the success/failure expectation
+			require.Equal(t, tt.expectSuccess, success, "validateWebsocketRequest success result mismatch")
+
+			// Verify context is not nil
+			require.NotNil(t, ctx, "returned context should not be nil")
+
+			// Additional verification based on expected result type
+			if tt.expectErrorContextType {
+				// For unsupported case, we expect some kind of error context
+				// We can't easily type assert the exact error context type without more imports,
+				// but we can verify it's not nil and success is false
+				require.False(t, success, "should return false for unsupported WebSocket")
+			} else {
+				// For supported case, we expect success
+				require.True(t, success, "should return true for supported WebSocket")
+			}
+		})
+	}
+}