[Proxy] feat: implement relayer proxy full workflow

go.mod

@@ -9,6 +9,7 @@ require (
 	cosmossdk.io/math v1.0.1
 	github.com/cometbft/cometbft v0.37.2
 	github.com/cometbft/cometbft-db v0.8.0
+	github.com/cosmos/cosmos-proto v1.0.0-beta.2
 	github.com/cosmos/cosmos-sdk v0.47.3
 	github.com/cosmos/gogoproto v1.4.10
 	github.com/cosmos/ibc-go/v7 v7.1.0
@@ -26,7 +27,9 @@ require (
 	github.com/stretchr/testify v1.8.4
 	go.uber.org/multierr v1.11.0
 	golang.org/x/crypto v0.12.0
+	golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df
 	golang.org/x/sync v0.3.0
+	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1
 	google.golang.org/grpc v1.56.1
 	gopkg.in/yaml.v2 v2.4.0
 )
@@ -70,7 +73,6 @@ require (
 	github.com/containerd/cgroups v1.1.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/cosmos/btcutil v1.0.5 // indirect
-	github.com/cosmos/cosmos-proto v1.0.0-beta.2 // indirect
 	github.com/cosmos/go-bip39 v1.0.0 // indirect
 	github.com/cosmos/gogogateway v1.2.0 // indirect
 	github.com/cosmos/iavl v0.20.0 // indirect
@@ -254,7 +256,6 @@ require (
 	go.uber.org/dig v1.16.1 // indirect
 	go.uber.org/fx v1.19.2 // indirect
 	go.uber.org/zap v1.24.0 // indirect
-	golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df // indirect
 	golang.org/x/mod v0.11.0 // indirect
 	golang.org/x/net v0.14.0 // indirect
 	golang.org/x/oauth2 v0.7.0 // indirect
@@ -266,7 +267,6 @@ require (
 	gonum.org/v1/gonum v0.11.0 // indirect
 	google.golang.org/api v0.122.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

pkg/relayer/proxy/interface.go → pkg/relayer/interface.go

@@ -1,4 +1,4 @@
-package proxy
+package relayer
 
 import (
 	"context"
@@ -27,10 +27,21 @@ type RelayerProxy interface {
 
 	// VerifyRelayRequest is a shared method used by RelayServers to check the
 	// relay request signature and session validity.
-	VerifyRelayRequest(relayRequest *types.RelayRequest) (isValid bool, err error)
-
-	// SignRelayResponse is a shared method used by RelayServers to sign the relay response.
-	SignRelayResponse(relayResponse *types.RelayResponse) ([]byte, error)
+	// This method may mutate the relay request in the future, so the returned
+	// relay request should be used instead of the passed in one.
+	// TODO_TECHDEBT(@red-0ne): This method should be moved out of the RelayerProxy interface
+	// that should not be responsible for verifying relay requests.
+	VerifyRelayRequest(
+		ctx context.Context,
+		relayRequest *types.RelayRequest,
+		serviceId *sharedtypes.ServiceId,
+	) (*types.RelayRequest, error)
+
+	// SignRelayResponse is a shared method used by RelayServers to sign
+	// and return a signed RelayResponse the relay response.
+	// TODO_TECHDEBT(@red-0ne): This method should be moved out of the RelayerProxy interface
+	// that should not be responsible for signing relay responses.
+	SignRelayResponse(relayResponse *types.RelayResponse) (*types.RelayResponse, error)
 }
 
 // RelayServer is the interface of the advertised relay servers provided by the RelayerProxy.

pkg/relayer/proxy/error_reply.go

@@ -0,0 +1,41 @@
+package proxy
+
+import (
+	"log"
+	"net/http"
+
+	"github.com/pokt-network/poktroll/x/service/types"
+)
+
+// replyWithError builds a JSONRPCResponseError from the passed in error and writes it to the writer.
+// TODO_TECHDEBT: This method should be aware of the request id and use it in the response by having
+// the caller pass it along with the error if available.
+// TODO_TECHDEBT: This method should be aware of the nature of the error to use the appropriate JSONRPC
+// Code, Message and Data. Possibly by augmenting the passed in error with the adequate information.
+func (j *jsonRPCServer) replyWithError(writer http.ResponseWriter, err error) {
+	relayResponse := &types.RelayResponse{
+		Payload: &types.RelayResponse_JsonRpcPayload{
+			JsonRpcPayload: &types.JSONRPCResponsePayload{
+				Id:      make([]byte, 0),
+				Jsonrpc: "2.0",
+				Error: &types.JSONRPCResponseError{
+					// Using conventional error code indicating internal server error.
+					Code:    -32000,
+					Message: err.Error(),
+					Data:    nil,
+				},
+			},
+		},
+	}
+
+	relayResponseBz, err := relayResponse.Marshal()
+	if err != nil {
+		log.Printf("ERROR: failed marshaling relay response: %s", err)
+		return
+	}
+
+	if _, err = writer.Write(relayResponseBz); err != nil {
+		log.Printf("ERROR: failed writing relay response: %s", err)
+		return
+	}
+}

pkg/relayer/proxy/errors.go

@@ -3,6 +3,9 @@ package proxy
 import sdkerrors "cosmossdk.io/errors"
 
 var (
-	codespace             = "relayer/proxy"
-	ErrUnsupportedRPCType = sdkerrors.Register(codespace, 1, "unsupported rpc type")
+	codespace                       = "relayer/proxy"
+	ErrUnsupportedRPCType           = sdkerrors.Register(codespace, 1, "unsupported rpc type")
+	ErrInvalidRelayRequestSignature = sdkerrors.Register(codespace, 2, "invalid relay request signature")
+	ErrInvalidSession               = sdkerrors.Register(codespace, 3, "invalid session")
+	ErrInvalidSupplier              = sdkerrors.Register(codespace, 4, "invalid supplier")
 )

pkg/relayer/proxy/jsonrpc.go

@@ -1,15 +1,19 @@
 package proxy
 
 import (
+	"bytes"
 	"context"
+	"io"
+	"log"
 	"net/http"
 	"net/url"
 
+	"github.com/pokt-network/poktroll/pkg/relayer"
 	"github.com/pokt-network/poktroll/x/service/types"
 	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
 )
 
-var _ RelayServer = (*jsonRPCServer)(nil)
+var _ relayer.RelayServer = (*jsonRPCServer)(nil)
 
 type jsonRPCServer struct {
 	// serviceId is the id of the service that the server is responsible for.
@@ -22,28 +26,28 @@ type jsonRPCServer struct {
 	// proxiedServiceEndpoint is the address of the proxied service that the server relays requests to.
 	proxiedServiceEndpoint url.URL
 
-	// server is the http server that listens for incoming relay requests.
+	// server is the HTTP server that listens for incoming relay requests.
 	server *http.Server
 
 	// relayerProxy is the main relayer proxy that the server uses to perform its operations.
-	relayerProxy RelayerProxy
+	relayerProxy relayer.RelayerProxy
 
-	// servedRelaysProducer is a channel that emits the relays that have been served so that the
-	// servedRelays observable can fan out the notifications to its subscribers.
+	// servedRelaysProducer is a channel that emits the relays that have been served, allowing
+	// the servedRelays observable to fan-out notifications to its subscribers.
 	servedRelaysProducer chan<- *types.Relay
 }
 
 // NewJSONRPCServer creates a new HTTP server that listens for incoming relay requests
 // and forwards them to the supported proxied service endpoint.
 // It takes the serviceId, endpointUrl, and the main RelayerProxy as arguments and returns
-// a RelayServer that listens to incoming RelayRequests
+// a RelayServer that listens to incoming RelayRequests.
 func NewJSONRPCServer(
 	serviceId *sharedtypes.ServiceId,
 	supplierEndpoint *sharedtypes.SupplierEndpoint,
 	proxiedServiceEndpoint url.URL,
 	servedRelaysProducer chan<- *types.Relay,
-	proxy RelayerProxy,
-) RelayServer {
+	proxy relayer.RelayerProxy,
+) relayer.RelayServer {
 	return &jsonRPCServer{
 		serviceId:              serviceId,
 		serverEndpoint:         supplierEndpoint,
@@ -57,29 +61,128 @@ func NewJSONRPCServer(
 // Start starts the service server and returns an error if it fails.
 // It also waits for the passed in context to end before shutting down.
 // This method is blocking and should be called in a goroutine.
-func (j *jsonRPCServer) Start(ctx context.Context) error {
+func (jsrv *jsonRPCServer) Start(ctx context.Context) error {
 	go func() {
 		<-ctx.Done()
-		j.server.Shutdown(ctx)
+		jsrv.server.Shutdown(ctx)
 	}()
 
-	return j.server.ListenAndServe()
+	return jsrv.server.ListenAndServe()
 }
 
 // Stop terminates the service server and returns an error if it fails.
-func (j *jsonRPCServer) Stop(ctx context.Context) error {
-	return j.server.Shutdown(ctx)
+func (jsrv *jsonRPCServer) Stop(ctx context.Context) error {
+	return jsrv.server.Shutdown(ctx)
 }
 
 // ServiceId returns the serviceId of the JSON-RPC service.
-func (j *jsonRPCServer) ServiceId() *sharedtypes.ServiceId {
-	return j.serviceId
+func (jsrv *jsonRPCServer) ServiceId() *sharedtypes.ServiceId {
+	return jsrv.serviceId
 }
 
 // ServeHTTP listens for incoming relay requests. It implements the respective
 // method of the http.Handler interface. It is called by http.ListenAndServe()
 // when jsonRPCServer is used as an http.Handler with an http.Server.
 // (see https://pkg.go.dev/net/http#Handler)
-func (j *jsonRPCServer) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
-	panic("TODO: implement jsonRPCServer.ServeHTTP")
+func (jsrv *jsonRPCServer) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	ctx := request.Context()
+	// Relay the request to the proxied service and build the response that will be sent back to the client.
+	relay, err := jsrv.serveHTTP(ctx, request)
+	if err != nil {
+		// Reply with an error if the relay could not be served.
+		jsrv.replyWithError(writer, err)
+		log.Printf("WARN: failed serving relay request: %s", err)
+		return
+	}
+
+	// Send the relay response to the client.
+	if err := jsrv.sendRelayResponse(relay.Res, writer); err != nil {
+		jsrv.replyWithError(writer, err)
+		log.Printf("WARN: failed sending relay response: %s", err)
+		return
+	}
+
+	log.Printf(
+		"INFO: relay request served successfully for application %s, service %s, session start block height %d, proxied service %s",
+		relay.Res.Meta.SessionHeader.ApplicationAddress,
+		relay.Res.Meta.SessionHeader.ServiceId.Id,
+		relay.Res.Meta.SessionHeader.SessionStartBlockHeight,
+		jsrv.serverEndpoint.Url,
+	)
+
+	// Emit the relay to the servedRelays observable.
+	jsrv.servedRelaysProducer <- relay
+}
+
+// serveHTTP holds the underlying logic of ServeHTTP.
+func (jsrv *jsonRPCServer) serveHTTP(ctx context.Context, request *http.Request) (*types.Relay, error) {
+	// Extract the relay request from the request body.
+	relayRequest, err := jsrv.newRelayRequest(request)
+	if err != nil {
+		return nil, err
+	}
+
+	// Verify the relay request signature and session.
+	// TODO_TECHDEBT(red-0ne): Currently, the relayer proxy is responsible for verifying
+	// the relay request signature. This responsibility should be shifted to the relayer itself.
+	// Consider using a middleware pattern to handle non-proxy specific logic, such as
+	// request signature verification, session verification, and response signature.
+	// This would help in separating concerns and improving code maintainability.
+	// See https://github.com/pokt-network/poktroll/issues/160
+	relayRequest, err = jsrv.relayerProxy.VerifyRelayRequest(ctx, relayRequest, jsrv.serviceId)
+	if err != nil {
+		return nil, err
+	}
+
+	// Get the relayRequest payload's `io.ReadCloser` to add it to the http.Request
+	// that will be sent to the proxied (i.e. staked for) service.
+	// (see https://pkg.go.dev/net/http#Request) Body field type.
+	var payloadBz []byte
+	if _, err = relayRequest.Payload.MarshalTo(payloadBz); err != nil {
+		return nil, err
+	}
+	requestBodyReader := io.NopCloser(bytes.NewBuffer(payloadBz))
+
+	// Build the request to be sent to the native service by substituting
+	// the destination URL's host with the native service's listen address.
+	destinationURL, err := url.Parse(request.URL.String())
+	if err != nil {
+		return nil, err
+	}
+	destinationURL.Host = jsrv.proxiedServiceEndpoint.Host
+
+	relayHTTPRequest := &http.Request{
+		Method: request.Method,
+		Header: request.Header,
+		URL:    destinationURL,
+		Host:   destinationURL.Host,
+		Body:   requestBodyReader,
+	}
+
+	// Send the relay request to the native service.
+	httpResponse, err := http.DefaultClient.Do(relayHTTPRequest)
+	if err != nil {
+		return nil, err
+	}
+
+	// Build the relay response from the native service response
+	// Use relayRequest.Meta.SessionHeader on the relayResponse session header since it was verified to be valid
+	// and has to be the same as the relayResponse session header.
+	relayResponse, err := jsrv.newRelayResponse(httpResponse, relayRequest.Meta.SessionHeader)
+	if err != nil {
+		return nil, err
+	}
+
+	return &types.Relay{Req: relayRequest, Res: relayResponse}, nil
+}
+
+// sendRelayResponse marshals the relay response and sends it to the client.
+func (j *jsonRPCServer) sendRelayResponse(relayResponse *types.RelayResponse, writer http.ResponseWriter) error {
+	relayResponseBz, err := relayResponse.Marshal()
+	if err != nil {
+		return err
+	}
+
+	_, err = writer.Write(relayResponseBz)
+	return err
 }