feat(rfq-relayer): relayer supports active quoting

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,83 @@ 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) {
+	chainIDs := []int{}
+	for chainID := range m.config.Chains {
+		chainIDs = append(chainIDs, chainID)
+	}
+	req := model.SubscribeActiveRFQRequest{
+		ChainIDs: 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)
+	}
+
+	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) {
+	if msg.Op != rest.RequestQuoteOp {
+		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)
+	}
+
+	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)
+	}
+
+	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,
+	}
+
+	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,10 @@ 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"`
+	// RFQWsURL is the URL of the RFQ websocket.
+	RFQWsURL *string `yaml:"rfq_ws_url"`
 	// RelayerAPIPort is the port of the relayer API.
 	RelayerAPIPort string `yaml:"relayer_api_port"`
 	// Database is the database config.

services/rfq/relayer/relconfig/getters.go

@@ -541,9 +541,14 @@ 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
+}
+
+// GetRFQWsURL returns the RFQ Ws URL.
+func (c Config) GetRFQWsURL() *string {
+	return c.RFQWsURL
 }
 
 // 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(), nil, sg)
+	apiClient, err := rfqAPIClient.NewAuthenticatedClient(metricHandler, cfg.GetRFQAPIURL(), cfg.GetRFQWsURL(), 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.GetRFQWsURL() != nil {
+		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 {