feat(rfq-relayer): relayer supports active quoting

services/rfq/api/client/client.go

@@ -230,6 +230,8 @@ func (c *clientImpl) connectWebsocket(ctx context.Context, req *model.SubscribeA
 	}
 
 	reqURL := strings.Replace(c.rClient.BaseURL, "http", "ws", 1) + rest.RFQStreamRoute
+	fmt.Printf("dialing websocket: %s\n", reqURL)
+	fmt.Printf("headers: %v\n", header)
 	conn, httpResp, err := websocket.DefaultDialer.Dial(reqURL, header)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to websocket: %w", err)

services/rfq/api/rest/server.go

@@ -275,7 +275,9 @@ func (r *QuoterAPIServer) AuthMiddleware() gin.HandlerFunc {
 				loggedRequest = &req
 			}
 		case RFQRoute, RFQStreamRoute:
+			fmt.Println("RFQRoute, RFQStreamRoute")
 			chainsHeader := c.GetHeader(ChainsHeader)
+			fmt.Printf("got chains header: %s\n", chainsHeader)
 			if chainsHeader != "" {
 				var chainIDs []int
 				err = json.Unmarshal([]byte(chainsHeader), &chainIDs)
@@ -434,28 +436,34 @@ func (r *QuoterAPIServer) PutRelayAck(c *gin.Context) {
 // @Header 101 {string} X-Api-Version "API Version Number - See docs for more info"
 // @Router /quote_requests [get].
 func (r *QuoterAPIServer) GetActiveRFQWebsocket(ctx context.Context, c *gin.Context) {
+	fmt.Println("GetActiveRFQWebsocket")
 	ctx, span := r.handler.Tracer().Start(ctx, "GetActiveRFQWebsocket")
 	defer func() {
 		metrics.EndSpan(span)
 	}()
 
+	fmt.Println("upgrading websocket")
 	ws, err := r.upgrader.Upgrade(c.Writer, c.Request, nil)
 	if err != nil {
 		logger.Error("Failed to set websocket upgrade", "error", err)
 		return
 	}
+	fmt.Println("upgraded websocket")
 
 	// use the relayer address as the ID for the connection
 	rawRelayerAddr, exists := c.Get("relayerAddr")
 	if !exists {
+		fmt.Println("no relayer address recovered from signature")
 		c.JSON(http.StatusBadRequest, gin.H{"error": "No relayer address recovered from signature"})
 		return
 	}
 	relayerAddr, ok := rawRelayerAddr.(string)
 	if !ok {
+		fmt.Println("invalid relayer address type")
 		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid relayer address type"})
 		return
 	}
+	fmt.Printf("relayer address: %s\n", relayerAddr)
 
 	span.SetAttributes(
 		attribute.String("relayer_address", relayerAddr),
@@ -464,6 +472,7 @@ func (r *QuoterAPIServer) GetActiveRFQWebsocket(ctx context.Context, c *gin.Cont
 	// only one connection per relayer allowed
 	_, ok = r.wsClients.Load(relayerAddr)
 	if ok {
+		fmt.Println("relayer already connected")
 		c.JSON(http.StatusBadRequest, gin.H{"error": "relayer already connected"})
 		return
 	}
@@ -474,12 +483,16 @@ func (r *QuoterAPIServer) GetActiveRFQWebsocket(ctx context.Context, c *gin.Cont
 	}()
 
 	client := newWsClient(relayerAddr, ws, r.pubSubManager, r.handler)
+	fmt.Println("registered ws client")
 	r.wsClients.Store(relayerAddr, client)
 	span.AddEvent("registered ws client")
+	fmt.Println("running ws client")
 	err = client.Run(ctx)
 	if err != nil {
+		fmt.Printf("error running ws client: %v\n", err)
 		logger.Error("Error running websocket client", "error", err)
 	}
+	fmt.Println("ws client done")
 }
 
 const (

services/rfq/api/rest/suite_test.go

@@ -44,7 +44,6 @@ type ServerSuite struct {
 	handler              metrics.Handler
 	QuoterAPIServer      *rest.QuoterAPIServer
 	port                 uint16
-	wsPort               uint16
 	originChainID        int
 	destChainID          int
 }
@@ -75,8 +74,6 @@ func (c *ServerSuite) SetupTest() {
 	c.True(ok)
 	port, err := freeport.GetFreePort()
 	c.port = uint16(port)
-	wsPort, err := freeport.GetFreePort()
-	c.wsPort = uint16(wsPort)
 	c.Require().NoError(err)
 
 	testConfig := config.Config{

services/rfq/api/rest/ws.go

@@ -96,6 +96,7 @@ const (
 
 // Run runs the WebSocket client.
 func (c *wsClient) Run(ctx context.Context) (err error) {
+	fmt.Println("running ws client")
 	c.lastPong = time.Now()
 	messageChan := make(chan []byte)
 	pingTicker := time.NewTicker(PingPeriod)
@@ -114,13 +115,17 @@ func (c *wsClient) Run(ctx context.Context) (err error) {
 			close(c.doneChan)
 			return nil
 		case req := <-c.requestChan:
+			fmt.Printf("sending quote request: %v\n", req)
 			err = c.sendRelayerRequest(ctx, req)
 			if err != nil {
+				fmt.Printf("error sending quote request: %v\n", err)
 				logger.Error("Error sending quote request: %s", err)
 			}
 		case msg := <-messageChan:
+			fmt.Printf("received message: %s\n", msg)
 			err = c.handleRelayerMessage(ctx, msg)
 			if err != nil {
+				fmt.Printf("error handling relayer message: %v\n", err)
 				logger.Error("Error handling relayer message: %s", err)
 			}
 		case <-pingTicker.C:
@@ -204,6 +209,7 @@ func (c *wsClient) handleRelayerMessage(ctx context.Context, msg []byte) (err er
 }
 
 func (c *wsClient) handleSubscribe(ctx context.Context, content json.RawMessage) (resp model.ActiveRFQMessage) {
+	fmt.Printf("handleSubscribe: %s\n", content)
 	ctx, span := c.handler.Tracer().Start(ctx, "handleSubscribe", trace.WithAttributes(
 		attribute.String("relayer_address", c.relayerAddr),
 	))
@@ -219,8 +225,10 @@ func (c *wsClient) handleSubscribe(ctx context.Context, content json.RawMessage)
 	span.SetAttributes(attribute.IntSlice("chain_ids", sub.Chains))
 	err = c.pubsub.AddSubscription(c.relayerAddr, sub)
 	if err != nil {
+		fmt.Printf("error adding subscription: %v\n", err)
 		return getErrorResponse(SubscribeOp, fmt.Errorf("error adding subscription: %w", err))
 	}
+	fmt.Printf("successfully added subscription for chain ids: %v\n", sub.Chains)
 	return getSuccessResponse(SubscribeOp)
 }
 

services/rfq/e2e/setup_test.go

@@ -314,7 +314,7 @@ func (i *IntegrationSuite) getRelayerConfig() relconfig.Config {
 		},
 		// generated ex-post facto
 		QuotableTokens: map[string][]string{},
-		RfqAPIURL:      i.apiServer,
+		RFQAPIURL:      i.apiServer,
 		Signer: signerConfig.SignerConfig{
 			Type: signerConfig.FileType.String(),
 			File: filet.TmpFile(i.T(), "", i.relayerWallet.PrivateKeyHex()).Name(),

services/rfq/relayer/quoter/quoter.go

@@ -3,6 +3,7 @@ package quoter
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"math/big"
@@ -31,6 +32,7 @@ import (
 	"github.com/synapsecns/sanguine/ethergo/signer/signer"
 	rfqAPIClient "github.com/synapsecns/sanguine/services/rfq/api/client"
 	"github.com/synapsecns/sanguine/services/rfq/api/model"
+	"github.com/synapsecns/sanguine/services/rfq/api/rest"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/inventory"
 )
 
@@ -42,6 +44,8 @@ var logger = log.Logger("quoter")
 type Quoter interface {
 	// SubmitAllQuotes submits all quotes to the RFQ API.
 	SubmitAllQuotes(ctx context.Context) (err error)
+	// SubscribeActiveRFQ subscribes to the RFQ websocket API.
+	SubscribeActiveRFQ(ctx context.Context) (err error)
 	// ShouldProcess determines if a quote should be processed.
 	// We do this by either saving all quotes in-memory, and refreshing via GetSelfQuotes() through the API
 	// The first comparison is does bridge transaction OriginChainID+TokenAddr match with a quote + DestChainID+DestTokenAddr, then we look to see if we have enough amount to relay it + if the price fits our bounds (based on that the Relayer is relaying the destination token for the origin)
@@ -251,6 +255,102 @@ func (m *Manager) SubmitAllQuotes(ctx context.Context) (err error) {
 	return m.prepareAndSubmitQuotes(ctx, inv)
 }
 
+// SubscribeActiveRFQ subscribes to the RFQ websocket API.
+// This function is blocking and will run until the context is cancelled.
+func (m *Manager) SubscribeActiveRFQ(ctx context.Context) (err error) {
+	ctx, span := m.metricsHandler.Tracer().Start(ctx, "SubscribeActiveRFQ")
+	defer func() {
+		metrics.EndSpanWithErr(span, err)
+	}()
+
+	chainIDs := []int{}
+	for chainID := range m.config.Chains {
+		chainIDs = append(chainIDs, chainID)
+	}
+	req := model.SubscribeActiveRFQRequest{
+		ChainIDs: chainIDs,
+	}
+	span.SetAttributes(attribute.IntSlice("chain_ids", chainIDs))
+
+	reqChan := make(chan *model.ActiveRFQMessage)
+	respChan, err := m.rfqClient.SubscribeActiveQuotes(ctx, &req, reqChan)
+	if err != nil {
+		return fmt.Errorf("error subscribing to active quotes: %w", err)
+	}
+	span.AddEvent("subscribed to active quotes")
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case msg := <-respChan:
+			resp, err := m.generateActiveRFQ(ctx, msg)
+			if err != nil {
+				return fmt.Errorf("error generating active RFQ message: %w", err)
+			}
+			respChan <- resp
+		}
+	}
+}
+
+// getActiveRFQ handles an active RFQ message.
+func (m *Manager) generateActiveRFQ(ctx context.Context, msg *model.ActiveRFQMessage) (resp *model.ActiveRFQMessage, err error) {
+	ctx, span := m.metricsHandler.Tracer().Start(ctx, "generateActiveRFQ", trace.WithAttributes(
+		attribute.String("op", msg.Op),
+	))
+	defer func() {
+		metrics.EndSpanWithErr(span, err)
+	}()
+
+	if msg.Op != rest.RequestQuoteOp {
+		span.AddEvent("not a request quote op")
+		return nil, nil
+	}
+
+	inv, err := m.inventoryManager.GetCommittableBalances(ctx, inventory.SkipDBCache())
+	if err != nil {
+		return nil, fmt.Errorf("error getting committable balances: %w", err)
+	}
+
+	var rfqRequest model.WsRFQRequest
+	err = json.Unmarshal(msg.Content, &rfqRequest)
+	if err != nil {
+		return nil, fmt.Errorf("error unmarshalling quote data: %w", err)
+	}
+	span.SetAttributes(attribute.String("request_id", rfqRequest.RequestID))
+
+	quoteInput := QuoteInput{
+		OriginChainID:   rfqRequest.Data.OriginChainID,
+		DestChainID:     rfqRequest.Data.DestChainID,
+		OriginTokenAddr: common.HexToAddress(rfqRequest.Data.OriginTokenAddr),
+		DestTokenAddr:   common.HexToAddress(rfqRequest.Data.DestTokenAddr),
+		OriginBalance:   inv[rfqRequest.Data.OriginChainID][common.HexToAddress(rfqRequest.Data.OriginTokenAddr)],
+		DestBalance:     inv[rfqRequest.Data.DestChainID][common.HexToAddress(rfqRequest.Data.DestTokenAddr)],
+	}
+
+	rawQuote, err := m.generateQuote(ctx, quoteInput)
+	if err != nil {
+		return nil, fmt.Errorf("error generating quote: %w", err)
+	}
+	span.SetAttributes(attribute.String("dest_amount", rawQuote.DestAmount))
+
+	rfqResp := model.WsRFQResponse{
+		RequestID:  rfqRequest.RequestID,
+		DestAmount: rawQuote.DestAmount,
+	}
+	respBytes, err := json.Marshal(rfqResp)
+	if err != nil {
+		return nil, fmt.Errorf("error serializing response: %w", err)
+	}
+	resp = &model.ActiveRFQMessage{
+		Op:      rest.SendQuoteOp,
+		Content: respBytes,
+	}
+	span.AddEvent("generated response")
+
+	return resp, nil
+}
+
 // GetPrice gets the price of a token.
 func (m *Manager) GetPrice(parentCtx context.Context, tokenName string) (_ float64, err error) {
 	ctx, span := m.metricsHandler.Tracer().Start(parentCtx, "GetPrice")

services/rfq/relayer/relconfig/config.go

@@ -33,8 +33,8 @@ type Config struct {
 	BaseChainConfig ChainConfig `yaml:"base_chain_config"`
 	// OmniRPCURL is the URL of the OmniRPC server.
 	OmniRPCURL string `yaml:"omnirpc_url"`
-	// RfqAPIURL is the URL of the RFQ API.
-	RfqAPIURL string `yaml:"rfq_url"`
+	// RFQAPIURL is the URL of the RFQ API.
+	RFQAPIURL string `yaml:"rfq_url"`
 	// RelayerAPIPort is the port of the relayer API.
 	RelayerAPIPort string `yaml:"relayer_api_port"`
 	// Database is the database config.
@@ -67,6 +67,8 @@ type Config struct {
 	SubmitSingleQuotes bool `yaml:"submit_single_quotes"`
 	// VolumeLimit is the maximum dollar value of relayed transactions in the BlockWindow.
 	VolumeLimit float64 `yaml:"volume_limit"`
+	// SupportActiveQuoting enables support for active quoting.
+	SupportActiveQuoting bool `yaml:"support_active_quoting"`
 }
 
 // ChainConfig represents the configuration for a chain.

services/rfq/relayer/relconfig/getters.go

@@ -541,9 +541,9 @@ func (c Config) GetOmniRPCURL() string {
 	return c.OmniRPCURL
 }
 
-// GetRfqAPIURL returns the RFQ API URL.
-func (c Config) GetRfqAPIURL() string {
-	return c.RfqAPIURL
+// GetRFQAPIURL returns the RFQ API URL.
+func (c Config) GetRFQAPIURL() string {
+	return c.RFQAPIURL
 }
 
 // GetDatabase returns the database config.

services/rfq/relayer/service/relayer.go

@@ -130,7 +130,7 @@ func NewRelayer(ctx context.Context, metricHandler metrics.Handler, cfg relconfi
 	priceFetcher := pricer.NewCoingeckoPriceFetcher(cfg.GetHTTPTimeout())
 	fp := pricer.NewFeePricer(cfg, omniClient, priceFetcher, metricHandler)
 
-	apiClient, err := rfqAPIClient.NewAuthenticatedClient(metricHandler, cfg.GetRfqAPIURL(), sg)
+	apiClient, err := rfqAPIClient.NewAuthenticatedClient(metricHandler, cfg.GetRFQAPIURL(), sg)
 	if err != nil {
 		return nil, fmt.Errorf("error creating RFQ API client: %w", err)
 	}
@@ -219,6 +219,16 @@ func (r *Relayer) Start(ctx context.Context) (err error) {
 		}
 	})
 
+	if r.cfg.SupportActiveQuoting {
+		g.Go(func() error {
+			err = r.quoter.SubscribeActiveRFQ(ctx)
+			if err != nil {
+				return fmt.Errorf("could not subscribe to active RFQ: %w", err)
+			}
+			return nil
+		})
+	}
+
 	g.Go(func() error {
 		for {
 			select {