synapsecns · dwasse · Sep 13, 2024 · Sep 13, 2024 · Sep 13, 2024 · Sep 13, 2024
@@ -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
+}
-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
-}
+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 conn.Close()
+ defer close(respChan)
+
+ 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
+}
-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
-}
+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 conn.Close()
+ defer close(respChan)
+
+ 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"

@@ -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,

@@ -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)
 }
 

@@ -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