feat(rfq): active quoting API

contrib/opbot/go.mod

@@ -209,6 +209,7 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.54.0 // indirect
 	github.com/prometheus/procfs v0.15.0 // indirect
+	github.com/puzpuzpuz/xsync v1.5.2 // indirect
 	github.com/puzpuzpuz/xsync/v2 v2.5.1 // indirect
 	github.com/richardwilkes/toolbox v1.74.0 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect

services/rfq/api/client/client.go

@@ -4,11 +4,15 @@
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"strconv"
+	"strings"
 	"time"
 
+	"github.com/ipfs/go-log"
+
 	"github.com/google/uuid"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 
@@ -17,17 +21,23 @@
 
 	"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")
+
+const pingPeriod = 20 * time.Second
+
 // 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 +47,7 @@
 	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)
+	PutRFQRequest(ctx context.Context, q *model.PutRFQRequest) (*model.PutRFQResponse, error)
 	resty() *resty.Client
 }
 
@@ -50,7 +61,8 @@
 
 type clientImpl struct {
 	UnauthenticatedClient
-	rClient *resty.Client
+	rClient   *resty.Client
+	reqSigner signer.Signer
 }
 
 // NewAuthenticatedClient creates a new client for the RFQ quoting API.
@@ -65,33 +77,40 @@
 	// 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,
+		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 +134,7 @@
 }
 
 // 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 +178,173 @@
 	return ack, nil
 }
 
+func (c *clientImpl) SubscribeActiveQuotes(ctx context.Context, req *model.SubscribeActiveRFQRequest, reqChan chan *model.ActiveRFQMessage) (respChan chan *model.ActiveRFQMessage, err error) {
+	conn, err := c.connectWebsocket(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to websocket: %w", err)
+	}
+
+	// first, subscrbe to the given chains
+	sub := model.SubscriptionParams{
+		Chains: req.ChainIDs,
+	}
+	subJSON, err := json.Marshal(sub)
+	if err != nil {
+		return respChan, fmt.Errorf("error marshaling subscription params: %w", err)
+	}
+	err = conn.WriteJSON(model.ActiveRFQMessage{
+		Op:      rest.SubscribeOp,
+		Content: json.RawMessage(subJSON),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error sending subscribe message: %w", err)
+	}
+
+	// make sure subscription is successful
+	var resp model.ActiveRFQMessage
+	err = conn.ReadJSON(&resp)
+	if err != nil {
+		return nil, fmt.Errorf("error reading subscribe response: %w", err)
+	}
+	if !resp.Success || resp.Op != rest.SubscribeOp {
+		return nil, fmt.Errorf("subscription failed")
+	}
+
+	respChan = make(chan *model.ActiveRFQMessage)
+	go func() {
+		err = c.processWebsocket(ctx, conn, reqChan, respChan)
+		if err != nil {
+			logger.Error("Error running websocket listener: %s", err)
+		}
+	}()
+
+	return respChan, nil
+}
+
+func (c *clientImpl) connectWebsocket(ctx context.Context, req *model.SubscribeActiveRFQRequest) (conn *websocket.Conn, err error) {
+	if len(req.ChainIDs) == 0 {
+		return nil, fmt.Errorf("chain IDs are required")
+	}
+
+	header, err := c.getWsHeaders(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get auth header: %w", err)
+	}
+
+	reqURL := strings.Replace(c.rClient.BaseURL, "http", "ws", 1) + rest.RFQStreamRoute
+	conn, httpResp, err := websocket.DefaultDialer.Dial(reqURL, header)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to websocket: %w", err)
+	}
+	err = httpResp.Body.Close()
+	if err != nil {
+		logger.Warnf("error closing websocket connection: %v", err)
+	}
+
+	return conn, nil
+}
+
+func (c *clientImpl) getWsHeaders(ctx context.Context, req *model.SubscribeActiveRFQRequest) (header http.Header, err error) {
+	header = http.Header{}
+	chainIDsJSON, err := json.Marshal(req.ChainIDs)
+	if err != nil {
+		return header, 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 header, fmt.Errorf("failed to get auth header: %w", err)
+	}
+	header.Set(rest.AuthorizationHeader, authHeader)
+	return header, nil
+}
+
+func (c *clientImpl) processWebsocket(ctx context.Context, conn *websocket.Conn, reqChan, respChan chan *model.ActiveRFQMessage) (err error) {
+	defer func() {
+		close(respChan)
+		err := conn.Close()
+		if err != nil {
+			logger.Warnf("error closing websocket connection: %v", err)
+		}
+	}()
+
+	readChan := make(chan []byte)
+	go c.listenWsMessages(ctx, conn, readChan)
+	go c.sendPings(ctx, reqChan)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case msg, ok := <-reqChan:
+			if !ok {
+				return fmt.Errorf("error reading from reqChan: %w", ctx.Err())
+			}
+			err := conn.WriteJSON(msg)
+			if err != nil {
+				return fmt.Errorf("error sending message to websocket: %w", err)
+			}
+		case msg, ok := <-readChan:
+			if !ok {
+				return nil
+			}
+			err = c.handleWsMessage(ctx, msg, respChan)
+			if err != nil {
+				return fmt.Errorf("error handling websocket message: %w", err)
+			}
+		}
+	}
+}
+
+func (c *clientImpl) sendPings(ctx context.Context, reqChan chan *model.ActiveRFQMessage) (err error) {
+	pingTicker := time.NewTicker(pingPeriod)
+	defer pingTicker.Stop()
+
+	for {
+		select {
+		case <-pingTicker.C:
+			pingMsg := model.ActiveRFQMessage{
+				Op: rest.PingOp,
+			}
+			reqChan <- &pingMsg
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+func (c *clientImpl) listenWsMessages(ctx context.Context, conn *websocket.Conn, readChan chan []byte) {
+	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
+		}
+		select {
+		case readChan <- message:
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (c *clientImpl) handleWsMessage(ctx context.Context, msg []byte, respChan chan *model.ActiveRFQMessage) (err error) {
+	var rfqMsg model.ActiveRFQMessage
+	err = json.Unmarshal(msg, &rfqMsg)
+	if err != nil {
+		return fmt.Errorf("error unmarshaling message: %w", err)
+	}
+
+	select {
+	case respChan <- &rfqMsg:
+	case <-ctx.Done():
+		return nil
+	}
+	return 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 +428,25 @@
 	return contracts, nil
 }
 
+func (c unauthenticatedClient) PutRFQRequest(ctx context.Context, q *model.PutRFQRequest) (*model.PutRFQResponse, error) {
+	var response model.PutRFQResponse
+	resp, err := c.rClient.R().
+		SetContext(ctx).
+		SetBody(q).
+		SetResult(&response).
+		Put(rest.RFQRoute)
+
+	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,