feat(rfq): active quoting API

services/rfq/api/client/client.go

@@ -4,11 +4,14 @@ package client
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"strconv"
 	"time"
 
+	"github.com/ipfs/go-log"
+
 	"github.com/google/uuid"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 
@@ -17,17 +20,21 @@ import (
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/go-resty/resty/v2"
+	"github.com/gorilla/websocket"
 	"github.com/synapsecns/sanguine/ethergo/signer/signer"
 	"github.com/synapsecns/sanguine/services/rfq/api/model"
 	"github.com/synapsecns/sanguine/services/rfq/api/rest"
 )
 
+var logger = log.Logger("rfq-client")
+
 // AuthenticatedClient is an interface for the RFQ API.
 // It provides methods for creating, retrieving and updating quotes.
 type AuthenticatedClient interface {
-	PutQuote(ctx context.Context, q *model.PutQuoteRequest) error
+	PutQuote(ctx context.Context, q *model.PutRelayerQuoteRequest) error
 	PutBulkQuotes(ctx context.Context, q *model.PutBulkQuotesRequest) error
 	PutRelayAck(ctx context.Context, req *model.PutAckRequest) (*model.PutRelayAckResponse, error)
+	SubscribeActiveQuotes(ctx context.Context, req *model.SubscribeActiveRFQRequest, reqChan chan *model.ActiveRFQMessage) (respChan chan *model.ActiveRFQMessage, err error)
 	UnauthenticatedClient
 }
 
@@ -37,6 +44,7 @@ type UnauthenticatedClient interface {
 	GetSpecificQuote(ctx context.Context, q *model.GetQuoteSpecificRequest) ([]*model.GetQuoteResponse, error)
 	GetQuoteByRelayerAddress(ctx context.Context, relayerAddr string) ([]*model.GetQuoteResponse, error)
 	GetRFQContracts(ctx context.Context) (*model.GetContractsResponse, error)
+	PutUserQuoteRequest(ctx context.Context, q *model.PutUserQuoteRequest) (*model.PutUserQuoteResponse, error)
 	resty() *resty.Client
 }
 
@@ -50,12 +58,14 @@ func (c unauthenticatedClient) resty() *resty.Client {
 
 type clientImpl struct {
 	UnauthenticatedClient
-	rClient *resty.Client
+	rClient   *resty.Client
+	wsURL     *string
+	reqSigner signer.Signer
 }
 
 // NewAuthenticatedClient creates a new client for the RFQ quoting API.
 // TODO: @aurelius,  you don't actually need to be authed for GET Requests.
-func NewAuthenticatedClient(metrics metrics.Handler, rfqURL string, reqSigner signer.Signer) (AuthenticatedClient, error) {
+func NewAuthenticatedClient(metrics metrics.Handler, rfqURL string, wsURL *string, reqSigner signer.Signer) (AuthenticatedClient, error) {
 	unauthedClient, err := NewUnauthenticatedClient(metrics, rfqURL)
 	if err != nil {
 		return nil, fmt.Errorf("could not create unauthenticated client: %w", err)
@@ -65,33 +75,41 @@ func NewAuthenticatedClient(metrics metrics.Handler, rfqURL string, reqSigner si
 	// to a new variable for clarity.
 	authedClient := unauthedClient.resty().
 		OnBeforeRequest(func(client *resty.Client, request *resty.Request) error {
-			// if request.Method == "PUT" && request.URL == rfqURL+rest.QUOTE_ROUTE {
-			// i.e. signature (hex encoded) = keccak(bytes.concat("\x19Ethereum Signed Message:\n", len(strconv.Itoa(time.Now().Unix()), strconv.Itoa(time.Now().Unix())))
-			// so that full auth header string: auth = strconv.Itoa(time.Now().Unix()) + ":" + signature
-			// Get the current Unix timestamp as a string.
-			now := strconv.Itoa(int(time.Now().Unix()))
-
-			// Prepare the data to be signed.
-			data := "\x19Ethereum Signed Message:\n" + strconv.Itoa(len(now)) + now
-
-			sig, err := reqSigner.SignMessage(request.Context(), []byte(data), true)
-
+			authHeader, err := getAuthHeader(request.Context(), reqSigner)
 			if err != nil {
-				return fmt.Errorf("failed to sign request: %w", err)
+				return fmt.Errorf("failed to get auth header: %w", err)
 			}
-
-			res := fmt.Sprintf("%s:%s", now, hexutil.Encode(signer.Encode(sig)))
-			request.SetHeader("Authorization", res)
-
+			request.SetHeader(rest.AuthorizationHeader, authHeader)
 			return nil
 		})
 
 	return &clientImpl{
 		UnauthenticatedClient: unauthedClient,
 		rClient:               authedClient,
+		wsURL:                 wsURL,
+		reqSigner:             reqSigner,
 	}, nil
 }
 
+func getAuthHeader(ctx context.Context, reqSigner signer.Signer) (string, error) {
+	// if request.Method == "PUT" && request.URL == rfqURL+rest.QUOTE_ROUTE {
+	// i.e. signature (hex encoded) = keccak(bytes.concat("\x19Ethereum Signed Message:\n", len(strconv.Itoa(time.Now().Unix()), strconv.Itoa(time.Now().Unix())))
+	// so that full auth header string: auth = strconv.Itoa(time.Now().Unix()) + ":" + signature
+	// Get the current Unix timestamp as a string.
+	now := strconv.Itoa(int(time.Now().Unix()))
+
+	// Prepare the data to be signed.
+	data := "\x19Ethereum Signed Message:\n" + strconv.Itoa(len(now)) + now
+
+	sig, err := reqSigner.SignMessage(ctx, []byte(data), true)
+
+	if err != nil {
+		return "", fmt.Errorf("failed to sign request: %w", err)
+	}
+
+	return fmt.Sprintf("%s:%s", now, hexutil.Encode(signer.Encode(sig))), nil
+}
+
 // NewUnauthenticatedClient creates a new client for the RFQ quoting API.
 func NewUnauthenticatedClient(metricHandler metrics.Handler, rfqURL string) (UnauthenticatedClient, error) {
 	client := resty.New().
@@ -115,7 +133,7 @@ func NewUnauthenticatedClient(metricHandler metrics.Handler, rfqURL string) (Una
 }
 
 // PutQuote puts a new quote in the RFQ quoting API.
-func (c *clientImpl) PutQuote(ctx context.Context, q *model.PutQuoteRequest) error {
+func (c *clientImpl) PutQuote(ctx context.Context, q *model.PutRelayerQuoteRequest) error {
 	res, err := c.rClient.R().
 		SetContext(ctx).
 		SetBody(q).
@@ -159,6 +177,91 @@ func (c *clientImpl) PutRelayAck(ctx context.Context, req *model.PutAckRequest)
 	return ack, nil
 }
 
+func (c *clientImpl) SubscribeActiveQuotes(ctx context.Context, req *model.SubscribeActiveRFQRequest, reqChan chan *model.ActiveRFQMessage) (respChan chan *model.ActiveRFQMessage, err error) {
+	if c.wsURL == nil {
+		return nil, fmt.Errorf("websocket URL is not set")
+	}
+	if len(req.ChainIDs) == 0 {
+		return nil, fmt.Errorf("chain IDs are required")
+	}
+
+	reqURL := *c.wsURL + rest.QuoteRequestsRoute
+
+	header := http.Header{}
+	chainIDsJSON, err := json.Marshal(req.ChainIDs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal chain IDs: %w", err)
+	}
+	header.Set(rest.ChainsHeader, string(chainIDsJSON))
+	authHeader, err := getAuthHeader(ctx, c.reqSigner)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get auth header: %w", err)
+	}
+	header.Set(rest.AuthorizationHeader, authHeader)
+
+	// Use the header when dialing
+	conn, _, err := websocket.DefaultDialer.Dial(reqURL, header)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to websocket: %w", err)
+	}
+
+	respChan = make(chan *model.ActiveRFQMessage, 1000)
+
+	go func() {
+		defer close(respChan)
+		defer conn.Close()
+
+		readChan := make(chan []byte, 1000)
+		go func() {
+			defer close(readChan)
+			for {
+				_, message, err := conn.ReadMessage()
+				if err != nil {
+					if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
+						logger.Warnf("websocket connection closed unexpectedly: %v", err)
+					}
+					return
+				}
+				readChan <- message
+			}
+		}()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case msg, ok := <-reqChan:
+				if !ok {
+					return
+				}
+				err := conn.WriteJSON(msg)
+				if err != nil {
+					logger.Warnf("error sending message to websocket: %v", err)
+					return
+				}
+			case msg, ok := <-readChan:
+				if !ok {
+					return
+				}
+				var rfqMsg model.ActiveRFQMessage
+				err = json.Unmarshal(msg, &rfqMsg)
+				if err != nil {
+					logger.Warn("error unmarshalling message: %v", err)
+					continue
+				}
+
+				select {
+				case respChan <- &rfqMsg:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	return respChan, nil
+}
+
 // GetAllQuotes retrieves all quotes from the RFQ quoting API.
 func (c *unauthenticatedClient) GetAllQuotes(ctx context.Context) ([]*model.GetQuoteResponse, error) {
 	var quotes []*model.GetQuoteResponse
@@ -242,6 +345,25 @@ func (c unauthenticatedClient) GetRFQContracts(ctx context.Context) (*model.GetC
 	return contracts, nil
 }
 
+func (c unauthenticatedClient) PutUserQuoteRequest(ctx context.Context, q *model.PutUserQuoteRequest) (*model.PutUserQuoteResponse, error) {
+	var response model.PutUserQuoteResponse
+	resp, err := c.rClient.R().
+		SetContext(ctx).
+		SetBody(q).
+		SetResult(&response).
+		Put(rest.PutQuoteRequestRoute)
+
+	if err != nil {
+		return nil, fmt.Errorf("error from server: %s: %w", getStatus(resp), err)
+	}
+
+	if resp.IsError() {
+		return nil, fmt.Errorf("error from server: %s", getStatus(resp))
+	}
+
+	return &response, nil
+}
+
 func getStatus(resp *resty.Response) string {
 	if resp == nil {
 		return "http status unavailable"

services/rfq/api/client/client_test.go

@@ -7,7 +7,7 @@ import (
 // TODO: @aurelius tese tests make a lot less sesnes w/ a composite index
 
 func (c *ClientSuite) TestPutAndGetQuote() {
-	req := model.PutQuoteRequest{
+	req := model.PutRelayerQuoteRequest{
 		OriginChainID:   1,
 		OriginTokenAddr: "0xOriginTokenAddr",
 		DestChainID:     42161,
@@ -40,7 +40,7 @@ func (c *ClientSuite) TestPutAndGetQuote() {
 
 func (c *ClientSuite) TestPutAndGetBulkQuotes() {
 	req := model.PutBulkQuotesRequest{
-		Quotes: []model.PutQuoteRequest{
+		Quotes: []model.PutRelayerQuoteRequest{
 			{
 				OriginChainID:   1,
 				OriginTokenAddr: "0xOriginTokenAddr",
@@ -98,7 +98,7 @@ func (c *ClientSuite) TestPutAndGetBulkQuotes() {
 }
 
 func (c *ClientSuite) TestGetSpecificQuote() {
-	req := model.PutQuoteRequest{
+	req := model.PutRelayerQuoteRequest{
 		OriginChainID:   1,
 		OriginTokenAddr: "0xOriginTokenAddr",
 		DestChainID:     42161,
@@ -135,7 +135,7 @@ func (c *ClientSuite) TestGetSpecificQuote() {
 }
 
 func (c *ClientSuite) TestGetQuoteByRelayerAddress() {
-	req := model.PutQuoteRequest{
+	req := model.PutRelayerQuoteRequest{
 		OriginChainID:   1,
 		OriginTokenAddr: "0xOriginTokenAddr",
 		DestChainID:     42161,

services/rfq/api/client/suite_test.go

@@ -103,7 +103,7 @@ func (c *ClientSuite) SetupTest() {
 	}()
 	time.Sleep(2 * time.Second) // Wait for the server to start.
 
-	c.client, err = client.NewAuthenticatedClient(metrics.Get(), fmt.Sprintf("http://127.0.0.1:%d", port), localsigner.NewSigner(c.testWallet.PrivateKey()))
+	c.client, err = client.NewAuthenticatedClient(metrics.Get(), fmt.Sprintf("http://127.0.0.1:%d", port), nil, localsigner.NewSigner(c.testWallet.PrivateKey()))
 	c.Require().NoError(err)
 }
 

services/rfq/api/config/config.go

@@ -26,6 +26,7 @@ type Config struct {
 	Port            string            `yaml:"port"`
 	RelayAckTimeout time.Duration     `yaml:"relay_ack_timeout"`
 	MaxQuoteAge     time.Duration     `yaml:"max_quote_age"`
+	WebsocketPort   *string           `yaml:"websocket_port"`
 }
 
 const defaultRelayAckTimeout = 30 * time.Second

services/rfq/api/model/request.go

@@ -1,7 +1,7 @@
 package model
 
-// PutQuoteRequest contains the schema for a PUT /quote request.
-type PutQuoteRequest struct {
+// PutRelayerQuoteRequest contains the schema for a PUT /quote request.
+type PutRelayerQuoteRequest struct {
 	OriginChainID           int    `json:"origin_chain_id"`
 	OriginTokenAddr         string `json:"origin_token_addr"`
 	DestChainID             int    `json:"dest_chain_id"`
@@ -15,7 +15,7 @@ type PutQuoteRequest struct {
 
 // PutBulkQuotesRequest contains the schema for a PUT /quote request.
 type PutBulkQuotesRequest struct {
-	Quotes []PutQuoteRequest `json:"quotes"`
+	Quotes []PutRelayerQuoteRequest `json:"quotes"`
 }
 
 // PutAckRequest contains the schema for a PUT /ack request.