feat(rfq-relayer): relayer supports active quoting

services/rfq/api/client/client.go

@@ -177,6 +177,7 @@ func (c *clientImpl) PutRelayAck(ctx context.Context, req *model.PutAckRequest)
 }
 
 func (c *clientImpl) SubscribeActiveQuotes(ctx context.Context, req *model.SubscribeActiveRFQRequest, reqChan chan *model.ActiveRFQMessage) (respChan chan *model.ActiveRFQMessage, err error) {
+	fmt.Println("client: subscribing")
 	conn, err := c.connectWebsocket(ctx, req)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to websocket: %w", err)
@@ -190,28 +191,34 @@ func (c *clientImpl) SubscribeActiveQuotes(ctx context.Context, req *model.Subsc
 	if err != nil {
 		return respChan, fmt.Errorf("error marshaling subscription params: %w", err)
 	}
+	fmt.Printf("%s sub: %v\n", time.Now().String(), sub)
 	err = conn.WriteJSON(model.ActiveRFQMessage{
 		Op:      rest.SubscribeOp,
 		Content: json.RawMessage(subJSON),
 	})
 	if err != nil {
+		fmt.Printf("%s error sending subscribe message: %v\n", time.Now().String(), err)
 		return nil, fmt.Errorf("error sending subscribe message: %w", err)
 	}
 
 	// make sure subscription is successful
+	fmt.Printf("%s waiting for subscribe response\n", time.Now().String())
 	var resp model.ActiveRFQMessage
 	err = conn.ReadJSON(&resp)
 	if err != nil {
 		return nil, fmt.Errorf("error reading subscribe response: %w", err)
 	}
+	fmt.Printf("%s subscribe response: %v\n", time.Now().String(), resp)
 	if !resp.Success || resp.Op != rest.SubscribeOp {
+		fmt.Printf("%s subscription failed: %v\n", time.Now().String(), resp)
 		return nil, fmt.Errorf("subscription failed")
 	}
 
 	respChan = make(chan *model.ActiveRFQMessage)
 	go func() {
 		err = c.processWebsocket(ctx, conn, reqChan, respChan)
 		if err != nil {
+			fmt.Printf("%s Error running websocket listener: %s\n", time.Now().String(), err)
 			logger.Error("Error running websocket listener: %s", err)
 		}
 	}()
@@ -220,6 +227,7 @@ func (c *clientImpl) SubscribeActiveQuotes(ctx context.Context, req *model.Subsc
 }
 
 func (c *clientImpl) connectWebsocket(ctx context.Context, req *model.SubscribeActiveRFQRequest) (conn *websocket.Conn, err error) {
+	fmt.Printf("%s connecting websocket: %v\n", time.Now().String(), req)
 	if len(req.ChainIDs) == 0 {
 		return nil, fmt.Errorf("chain IDs are required")
 	}
@@ -230,15 +238,19 @@ 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)
 	}
 	err = httpResp.Body.Close()
 	if err != nil {
+		fmt.Printf("error closing resp body: %v\n", err)
 		logger.Warnf("error closing websocket connection: %v", err)
 	}
 
+	fmt.Println("connected to websocket")
 	return conn, nil
 }
 
@@ -258,10 +270,12 @@ func (c *clientImpl) getWsHeaders(ctx context.Context, req *model.SubscribeActiv
 }
 
 func (c *clientImpl) processWebsocket(ctx context.Context, conn *websocket.Conn, reqChan, respChan chan *model.ActiveRFQMessage) (err error) {
+	fmt.Printf("%s processing websocket\n", time.Now().String())
 	defer func() {
 		close(respChan)
 		err := conn.Close()
 		if err != nil {
+			fmt.Printf("%s Error closing websocket connection: %v\n", time.Now().String(), err)
 			logger.Warnf("error closing websocket connection: %v", err)
 		}
 	}()
@@ -272,9 +286,11 @@ func (c *clientImpl) processWebsocket(ctx context.Context, conn *websocket.Conn,
 	for {
 		select {
 		case <-ctx.Done():
+			fmt.Printf("%s context done for processWebsocket\n", time.Now().String())
 			return nil
 		case msg, ok := <-reqChan:
 			if !ok {
+				fmt.Printf("%s reqChan closed\n", time.Now().String())
 				return nil
 			}
 			err := conn.WriteJSON(msg)
@@ -283,10 +299,12 @@ func (c *clientImpl) processWebsocket(ctx context.Context, conn *websocket.Conn,
 			}
 		case msg, ok := <-readChan:
 			if !ok {
+				fmt.Printf("%s readChan closed\n", time.Now().String())
 				return nil
 			}
 			err = c.handleWsMessage(ctx, msg, reqChan, respChan)
 			if err != nil {
+				fmt.Printf("%s error handling websocket message: %v\n", time.Now().String(), err)
 				return fmt.Errorf("error handling websocket message: %w", err)
 			}
 		}
@@ -297,6 +315,8 @@ func (c *clientImpl) listenWsMessages(ctx context.Context, conn *websocket.Conn,
 	defer close(readChan)
 	for {
 		_, message, err := conn.ReadMessage()
+		fmt.Printf("read message: %v\n", message)
+		fmt.Printf("read message err: %v\n", err)
 		if err != nil {
 			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
 				logger.Warnf("websocket connection closed unexpectedly: %v", err)
@@ -306,15 +326,18 @@ func (c *clientImpl) listenWsMessages(ctx context.Context, conn *websocket.Conn,
 		select {
 		case readChan <- message:
 		case <-ctx.Done():
+			fmt.Println("context done for listenWsMessages")
 			return
 		}
 	}
 }
 
 func (c *clientImpl) handleWsMessage(ctx context.Context, msg []byte, reqChan, respChan chan *model.ActiveRFQMessage) (err error) {
+	fmt.Printf("%s handling websocket message: %v\n", time.Now().String(), msg)
 	var rfqMsg model.ActiveRFQMessage
 	err = json.Unmarshal(msg, &rfqMsg)
 	if err != nil {
+		fmt.Printf("%s error unmarshaling message: %v\n", time.Now().String(), err)
 		return fmt.Errorf("error unmarshaling message: %w", err)
 	}
 
@@ -333,7 +356,9 @@ func (c *clientImpl) handleWsMessage(ctx context.Context, msg []byte, reqChan, r
 
 	select {
 	case respChan <- &rfqMsg:
+		fmt.Printf("%s sent message to respChan\n", time.Now().String())
 	case <-ctx.Done():
+		fmt.Printf("%s context done for handleWsMessage\n", time.Now().String())
 		return nil
 	}
 	return nil

services/rfq/api/db/api_db.go

@@ -153,8 +153,8 @@ type ActiveQuoteRequest struct {
 	ExpirationWindow time.Duration            `gorm:"column:expiration_window"`
 	CreatedAt        time.Time                `gorm:"column:created_at"`
 	Status           ActiveQuoteRequestStatus `gorm:"column:status"`
-	ClosedAt         time.Time                `gorm:"column:fulfilled_at"`
-	ClosedQuoteID    string                   `gorm:"column:fulfilled_quote_id"`
+	ClosedAt         *time.Time               `gorm:"column:closed_at"`
+	ClosedQuoteID    *string                  `gorm:"column:closed_quote_id"`
 }
 
 // FromUserRequest converts a model.PutRFQRequest to an ActiveQuoteRequest.

services/rfq/api/db/sql/base/store.go

@@ -102,8 +102,8 @@ func (s *Store) UpdateActiveQuoteRequestStatus(ctx context.Context, requestID st
 		if quoteID == nil {
 			return fmt.Errorf("quote id is required for fulfilled status")
 		}
-		updates["fulfilled_quote_id"] = quoteID
-		updates["fulfilled_at"] = time.Now().UTC()
+		updates["closed_quote_id"] = quoteID
+		updates["closed_at"] = time.Now().UTC()
 	}
 	result := s.db.WithContext(ctx).
 		Model(&db.ActiveQuoteRequest{}).

services/rfq/api/model/response.go

@@ -56,17 +56,17 @@ type ActiveRFQMessage struct {
 
 // PutUserQuoteResponse represents a response to a user quote request.
 type PutUserQuoteResponse struct {
-	Success     bool      `json:"success"`
-	Reason      string    `json:"reason"`
-	UserAddress string    `json:"user_address"`
-	QuoteType   string    `json:"quote_type"`
-	Data        QuoteData `json:"data"`
+	Success        bool   `json:"success"`
+	Reason         string `json:"reason,omitempty"`
+	QuoteType      string `json:"quote_type,omitempty"`
+	DestAmount     string `json:"dest_amount,omitempty"`
+	RelayerAddress string `json:"relayer_address,omitempty"`
 }
 
 // WsRFQResponse represents a response to a quote request.
 type WsRFQResponse struct {
 	RequestID  string    `json:"request_id"`
-	QuoteID    string    `json:"quote_id"`
+	QuoteID    string    `json:"quote_id,omitempty"`
 	DestAmount string    `json:"dest_amount"`
 	UpdatedAt  time.Time `json:"updated_at"`
 }

services/rfq/api/rest/rfq.go

@@ -55,11 +55,12 @@ func (r *QuoterAPIServer) handleActiveRFQ(ctx context.Context, request *model.Pu
 	responses := r.collectRelayerResponses(ctx, request, requestID)
 	var quoteID string
 	var isUpdated bool
-	for _, resp := range responses {
+	for relayerAddr, resp := range responses {
 		quote, isUpdated = getBestQuote(quote, getRelayerQuoteData(request, resp))
 		if isUpdated {
 			quoteID = resp.QuoteID
 		}
+		quote.RelayerAddress = &relayerAddr
 	}
 	err = r.recordActiveQuote(ctx, quote, requestID, quoteID)
 	if err != nil {
@@ -90,13 +91,14 @@ func (r *QuoterAPIServer) collectRelayerResponses(ctx context.Context, request *
 		wg.Add(1)
 		go func(client WsClient) {
 			var respStatus db.ActiveQuoteResponseStatus
+			var err error
 			_, clientSpan := r.handler.Tracer().Start(collectionCtx, "collectRelayerResponses", trace.WithAttributes(
 				attribute.String("relayer_address", relayerAddr),
 				attribute.String("request_id", requestID),
 			))
 			defer func() {
 				clientSpan.SetAttributes(attribute.String("status", respStatus.String()))
-				metrics.EndSpan(clientSpan)
+				metrics.EndSpanWithErr(clientSpan, err)
 			}()
 
 			defer wg.Done()
@@ -105,6 +107,11 @@ func (r *QuoterAPIServer) collectRelayerResponses(ctx context.Context, request *
 				logger.Errorf("Error receiving quote response: %v", err)
 				return
 			}
+			span.AddEvent("received quote response", trace.WithAttributes(
+				attribute.String("relayer_address", relayerAddr),
+				attribute.String("request_id", requestID),
+				attribute.String("dest_amount", resp.DestAmount),
+			))
 
 			// validate the response
 			respStatus = getQuoteResponseStatus(expireCtx, resp)

services/rfq/api/rest/rfq_test.go

@@ -103,7 +103,6 @@ func (c *ServerSuite) TestActiveRFQSingleRelayer() {
 	}
 
 	// Prepare the relayer quote response
-	originAmount := userRequestAmount.String()
 	destAmount := new(big.Int).Sub(userRequestAmount, big.NewInt(1000)).String()
 	quoteResp := &model.WsRFQResponse{
 		DestAmount: destAmount,
@@ -119,8 +118,7 @@ func (c *ServerSuite) TestActiveRFQSingleRelayer() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("active", userQuoteResp.QuoteType)
-	c.Assert().Equal(destAmount, *userQuoteResp.Data.DestAmount)
-	c.Assert().Equal(originAmount, userQuoteResp.Data.OriginAmount)
+	c.Assert().Equal(destAmount, userQuoteResp.DestAmount)
 
 	// Verify ActiveQuoteRequest insertion
 	activeQuoteRequests, err := c.database.GetActiveQuoteRequests(c.GetTestContext())
@@ -206,7 +204,6 @@ func (c *ServerSuite) TestActiveRFQMultipleRelayers() {
 	}
 
 	// Prepare the relayer quote responses
-	originAmount := userRequestAmount.String()
 	destAmount := "999000"
 	quoteResp := model.WsRFQResponse{
 		DestAmount: destAmount,
@@ -234,8 +231,7 @@ func (c *ServerSuite) TestActiveRFQMultipleRelayers() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("active", userQuoteResp.QuoteType)
-	c.Assert().Equal(destAmount, *userQuoteResp.Data.DestAmount)
-	c.Assert().Equal(originAmount, userQuoteResp.Data.OriginAmount)
+	c.Assert().Equal(destAmount, userQuoteResp.DestAmount)
 
 	// Verify ActiveQuoteRequest insertion
 	activeQuoteRequests, err := c.database.GetActiveQuoteRequests(c.GetTestContext())
@@ -309,9 +305,8 @@ func (c *ServerSuite) TestActiveRFQFallbackToPassive() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("passive", userQuoteResp.QuoteType)
-	c.Assert().Equal("998000", *userQuoteResp.Data.DestAmount) // destAmount is quote destAmount minus fixed fee
-	c.Assert().Equal(userRequestAmount.String(), userQuoteResp.Data.OriginAmount)
-	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), *userQuoteResp.Data.RelayerAddress)
+	c.Assert().Equal("998000", userQuoteResp.DestAmount) // destAmount is quote destAmount minus fixed fee
+	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), userQuoteResp.RelayerAddress)
 }
 
 func (c *ServerSuite) TestActiveRFQPassiveBestQuote() {
@@ -389,7 +384,6 @@ func (c *ServerSuite) TestActiveRFQPassiveBestQuote() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("passive", userQuoteResp.QuoteType)
-	c.Assert().Equal("998900", *userQuoteResp.Data.DestAmount) // destAmount is quote destAmount minus fixed fee
-	c.Assert().Equal(userRequestAmount.String(), userQuoteResp.Data.OriginAmount)
-	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), *userQuoteResp.Data.RelayerAddress)
+	c.Assert().Equal("998900", userQuoteResp.DestAmount) // destAmount is quote destAmount minus fixed fee
+	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), userQuoteResp.RelayerAddress)
 }