Merge remote-tracking branch 'upstream/develop' into develop

go.mod

@@ -11,11 +11,12 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/websocket v1.5.3
+	github.com/jellydator/ttlcache/v3 v3.3.0
 	github.com/sourcegraph/jsonrpc2 v0.2.0
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/viper v1.19.0
 	github.com/valyala/fastjson v1.6.4
-	golang.org/x/sync v0.7.0
+	golang.org/x/sync v0.8.0
 	google.golang.org/grpc v1.64.0
 )
 

go.sum

@@ -117,6 +117,8 @@ github.com/holiman/uint256 v1.3.1 h1:JfTzmih28bittyHM8z360dCjIA9dbPIBlcTI6lmctQs
 github.com/holiman/uint256 v1.3.1/go.mod h1:EOMSn4q6Nyt9P6efbI3bueV4e1b3dGlUCXeiRV4ng7E=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
 github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM=
@@ -244,6 +246,8 @@ github.com/valyala/fastjson v1.6.4 h1:uAUNq9Z6ymTgGhcm0UynUAB6tlbakBrz6CQFax3BXV
 github.com/valyala/fastjson v1.6.4/go.mod h1:CLCAqky6SMuOcxStkYQvblddUtoRxhYMGLrsQns1aXY=
 github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw=
 github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
@@ -252,8 +256,8 @@ golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 h1:vr/HnozRka3pE4EsMEg1lgkXJ
 golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842/go.mod h1:XtvwrStGgqGPLc4cjQfWqZHG1YFdYs6swckp8vpsjnc=
 golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac=
 golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
-golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
-golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
+golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

relay/relay.go

@@ -23,7 +23,7 @@ func Run(cfg *config.Config) error {
 	}
 
 	eg.Go(func() error {
-		return s.Start()
+		return s.Start(gCtx)
 	})
 
 	<-gCtx.Done()

relay/server/dapp.go

@@ -40,6 +40,8 @@ func (s *Server) userOperation(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	s.intentService.SubscribeToIntentSolutions(intentID)
+
 	writeResponseData(w, map[string]string{
 		"intent_id": intentID,
 	})

relay/server/server.go

@@ -37,6 +37,11 @@ func NewServer(ctx context.Context, cfg *config.Config) (*Server, error) {
 		return nil, fmt.Errorf("failed to subscribe to intents: %v", err)
 	}
 
+	err = intentService.SubscribeToSolutions(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to subscribe to solutions: %v", err)
+	}
+
 	return &Server{
 		cfg:                 cfg,
 		intentService:       intentService,
@@ -45,7 +50,13 @@ func NewServer(ctx context.Context, cfg *config.Config) (*Server, error) {
 }
 
 // Start setup handlers and start http server
-func (s *Server) Start() error {
+func (s *Server) Start(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return nil
+	default:
+	}
+
 	s.server = &http.Server{
 		Addr:              fmt.Sprintf(":%v", s.cfg.HTTPPort),
 		ReadHeaderTimeout: time.Second * 5,

relay/service/intent.go

@@ -11,6 +11,7 @@ import (
 	"github.com/FastLane-Labs/atlas-sdk-go/types"
 	sdk "github.com/bloXroute-Labs/bloxroute-sdk-go"
 	"github.com/bloXroute-Labs/bloxroute-sdk-go/connection/ws"
+	"github.com/jellydator/ttlcache/v3"
 	"github.com/valyala/fastjson"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -24,6 +25,7 @@ type Intent struct {
 	client              *sdk.Client
 	cfg                 *config.Config
 	subscriptionManager *SubscriptionManager
+	cache               *ttlcache.Cache[string, []types.SolverOperationRaw]
 }
 
 // NewIntent creates a new Intent service
@@ -53,10 +55,17 @@ func NewIntent(ctx context.Context, cfg *config.Config, subscriptionManager *Sub
 		return nil, fmt.Errorf("failed to create BDN client: %w", err)
 	}
 
+	cache := ttlcache.New[string, []types.SolverOperationRaw](
+		ttlcache.WithTTL[string, []types.SolverOperationRaw](time.Minute),
+	)
+
+	go cache.Start()
+
 	return &Intent{
 		client:              client,
 		cfg:                 cfg,
 		subscriptionManager: subscriptionManager,
+		cache:               cache,
 	}, nil
 }
 
@@ -88,6 +97,8 @@ func (i *Intent) SubmitIntent(ctx context.Context, intent []byte) (string, error
 }
 
 func (i *Intent) SubscribeToIntents(ctx context.Context) error {
+	logger.Debug("subscribing to intents")
+
 	params := &sdk.IntentsParams{
 		SolverPrivateKey: i.cfg.SolverPrivateKey,
 		// TODO uncomment when the BDN supports filtering by DApp address
@@ -141,6 +152,13 @@ func (i *Intent) SubmitIntentSolution(ctx context.Context, intentID string, inte
 
 // GetIntentSolutions gets list of solutions for a specific intent
 func (i *Intent) GetIntentSolutions(ctx context.Context, intentID string) ([]types.SolverOperationRaw, error) {
+	// check if we have the solutions in cache
+	item := i.cache.Get(intentID)
+	if item != nil && len(item.Value()) != 0 {
+		logger.Debug("returning cached intent solutions", "intent_id", intentID)
+		return item.Value(), nil
+	}
+
 	params := &sdk.GetSolutionsForIntentParams{
 		DAppOrSenderPrivateKey: i.cfg.DAppPrivateKey,
 		IntentID:               intentID,
@@ -183,3 +201,55 @@ func (i *Intent) GetIntentSolutions(ctx context.Context, intentID string) ([]typ
 
 	return result, nil
 }
+
+func (i *Intent) SubscribeToSolutions(ctx context.Context) error {
+	logger.Debug("subscribing to intent solutions")
+
+	params := &sdk.IntentSolutionsParams{
+		DappPrivateKey: i.cfg.DAppPrivateKey,
+	}
+
+	err := i.client.OnIntentSolutions(ctx, params, func(ctx context.Context, err error, result *sdk.OnIntentSolutionsNotification) {
+		if err != nil {
+			logger.Error("error receiving intent solution", "error", err)
+			return
+		}
+		logger.Debug("received intent solution", "intent_id", result.IntentID)
+
+		item := i.cache.Get(result.IntentID)
+		if item == nil {
+			return
+		}
+
+		v := item.Value()
+
+		out := make([]byte, base64.StdEncoding.DecodedLen(len(result.IntentSolution)))
+		n, err := base64.StdEncoding.Decode(out, result.IntentSolution)
+		if err != nil {
+			logger.Error("failed to decode intent solution from base64", "error", err, "intent_solution", string(result.IntentSolution))
+			return
+		}
+
+		var solverOperation *types.SolverOperationRaw
+		err = json.Unmarshal(out[:n], &solverOperation)
+		if err != nil {
+			logger.Error("failed to unmarshal intent solution into SolverOperationRaw", "error", err,
+				"intent_solution", string(result.IntentSolution))
+			return
+		}
+
+		v = append(v, *solverOperation)
+
+		i.cache.Set(result.IntentID, v, ttlcache.DefaultTTL)
+	})
+
+	if err != nil {
+		return fmt.Errorf("failed to subscribe to intent solutions: %w", err)
+	}
+
+	return nil
+}
+
+func (i *Intent) SubscribeToIntentSolutions(intentID string) {
+	i.cache.Set(intentID, []types.SolverOperationRaw{}, ttlcache.DefaultTTL)
+}