Order updates websockets

data/redisrepo/pub_sub.go

@@ -0,0 +1,53 @@
+package redisrepo
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/orbs-network/order-book/utils/logger"
+	"github.com/orbs-network/order-book/utils/logger/logctx"
+)
+
+// PublishEvent publishes an event to Redis
+func (r *redisRepository) PublishEvent(ctx context.Context, key string, value interface{}) error {
+	err := r.client.Publish(ctx, key, value).Err()
+
+	if err != nil {
+		logctx.Error(ctx, "failed to publish redis event", logger.Error(err), logger.String("key", key))
+		return fmt.Errorf("failed to publish redis event: %v", err)
+	}
+
+	logctx.Info(ctx, "published redis event", logger.String("key", key))
+	return nil
+}
+
+// SubscribeToEvents subscribes to events on a given Redis channel
+func (r *redisRepository) SubscribeToEvents(ctx context.Context, channel string) (chan []byte, error) {
+	logctx.Info(ctx, "subscribing to channel", logger.String("channel", channel))
+
+	// Subscribe to the specified channel
+	pubsub := r.client.Subscribe(ctx, channel)
+
+	// Wait for confirmation that subscription is created
+	_, err := pubsub.Receive(ctx)
+	if err != nil {
+		logctx.Error(ctx, "error on receiving from pubsub", logger.Error(err), logger.String("channel", channel))
+		return nil, fmt.Errorf("error on receiving from pubsub: %w", err)
+	}
+
+	// Create a channel to pass messages to the caller
+	messages := make(chan []byte)
+
+	// Listen for messages
+	go func() {
+		defer pubsub.Close()
+		ch := pubsub.Channel()
+		for msg := range ch {
+			messages <- []byte(msg.Payload)
+		}
+		logctx.Info(ctx, "subscription ended", logger.String("channel", channel))
+		close(messages)
+	}()
+
+	return messages, nil
+}

data/redisrepo/pub_sub_test.go

@@ -0,0 +1,41 @@
+package redisrepo
+
+import (
+	"testing"
+
+	"github.com/go-redis/redismock/v9"
+	"github.com/stretchr/testify/assert"
+)
+
+var orderJson, _ = order.ToJson()
+
+func TestRedisRepository_PublishEvent(t *testing.T) {
+
+	t.Run("should publish event", func(t *testing.T) {
+		db, mock := redismock.NewClientMock()
+
+		repo := &redisRepository{
+			client: db,
+		}
+
+		mock.ExpectPublish(order.Id.String(), orderJson).SetVal(1)
+
+		err := repo.PublishEvent(ctx, order.Id.String(), orderJson)
+
+		assert.NoError(t, err)
+	})
+
+	t.Run("should return error when failed to publish event", func(t *testing.T) {
+		db, mock := redismock.NewClientMock()
+
+		repo := &redisRepository{
+			client: db,
+		}
+
+		mock.ExpectPublish(order.Id.String(), orderJson).SetErr(assert.AnError)
+
+		err := repo.PublishEvent(ctx, order.Id.String(), orderJson)
+
+		assert.ErrorContains(t, err, "failed to publish redis event")
+	})
+}

data/redisrepo/repository.go

@@ -7,18 +7,26 @@ import (
 )
 
 type redisRepository struct {
-	client  redis.Cmdable
+	cmdable redis.Cmdable
+	client  *redis.Client
 	txMap   map[uint]redis.Pipeliner
 	ixIndex uint
 }
 
-func NewRedisRepository(client redis.Cmdable) (*redisRepository, error) {
-	if client == nil {
+func NewRedisRepository(cmdable redis.Cmdable) (*redisRepository, error) {
+	if cmdable == nil {
 		return nil, fmt.Errorf("redis client cannot be nil")
 	}
+
+	client, ok := cmdable.(*redis.Client)
+	if !ok {
+		return nil, fmt.Errorf("cmdable is not a *redis.Client")
+	}
+
 	txMap := make(map[uint]redis.Pipeliner)
 	return &redisRepository{
-		client: client,
-		txMap:  txMap,
+		cmdable: cmdable,
+		client:  client,
+		txMap:   txMap,
 	}, nil
 }

data/store/store.go

@@ -72,4 +72,8 @@ type OrderBookStore interface {
 	StoreUserResolvedSwap(ctx context.Context, userId uuid.UUID, swap models.Swap) error
 	// utils
 	EnumSubKeysOf(ctx context.Context, key string) ([]string, error)
+
+	// PubSub
+	PublishEvent(ctx context.Context, key string, value interface{}) error
+	SubscribeToEvents(ctx context.Context, channel string) (chan []byte, error)
 }

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/go-redis/redismock/v9 v9.2.0
 	github.com/google/uuid v1.3.1
 	github.com/gorilla/mux v1.8.0
+	github.com/gorilla/websocket v1.4.2
 	github.com/redis/go-redis/v9 v9.2.1
 	github.com/shopspring/decimal v1.3.1
 	github.com/stretchr/testify v1.8.4
@@ -46,7 +47,6 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
-	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
 	github.com/holiman/uint256 v1.2.3 // indirect
 	github.com/huin/goupnp v1.3.0 // indirect

mocks/service.go

@@ -27,6 +27,8 @@ type MockOrderBookStore struct {
 	Sets map[string]map[string]struct{}
 	// Pending swaps
 	PendingSwaps []models.SwapTx
+	// PubSub
+	EventsChan chan []byte
 }
 
 func (m *MockOrderBookStore) StoreOpenOrder(ctx context.Context, order models.Order) error {
@@ -188,3 +190,11 @@ func (m *MockOrderBookStore) EnumSubKeysOf(tx context.Context, key string) ([]st
 func (m *MockOrderBookStore) CancelPendingOrder(ctx context.Context, order models.Order) error {
 	return m.Error
 }
+
+func (m *MockOrderBookStore) PublishEvent(ctx context.Context, key string, value interface{}) error {
+	return m.Error
+}
+
+func (m *MockOrderBookStore) SubscribeToEvents(ctx context.Context, channel string) (chan []byte, error) {
+	return m.EventsChan, m.Error
+}

mocks/transport.go

@@ -19,6 +19,7 @@ type MockOrderBookService struct {
 	Symbols      []models.Symbol
 	User         *models.User
 	BeginSwapRes models.BeginSwapRes
+	OrderEvents  chan []byte
 }
 
 func (m *MockOrderBookService) GetUserByPublicKey(ctx context.Context, publicKey string) (*models.User, error) {
@@ -73,6 +74,10 @@ func (m *MockOrderBookService) CancelOrdersForUser(ctx context.Context, userId u
 	return ids, m.Error
 }
 
+func (m *MockOrderBookService) SubscribeUserOrders(ctx context.Context, userId uuid.UUID) (chan []byte, error) {
+	return m.OrderEvents, m.Error
+}
+
 func (m *MockOrderBookService) GetQuote(ctx context.Context, symbol models.Symbol, side models.Side, inAmount decimal.Decimal, minOutAmount *decimal.Decimal) (models.QuoteRes, error) {
 	return m.QuoteRes, m.Error
 }

models/events.go

@@ -0,0 +1,13 @@
+// Pub/Sub events
+
+package models
+
+import (
+	"fmt"
+
+	"github.com/google/uuid"
+)
+
+func CreateUserOrdersEventKey(userId uuid.UUID) string {
+	return fmt.Sprintf("user_orders:%s", userId)
+}

models/order.go

@@ -207,6 +207,10 @@ func (o *Order) MapToOrder(data map[string]string) error {
 	return nil
 }
 
+func (o *Order) ToJson() ([]byte, error) {
+	return json.Marshal(o)
+}
+
 // GetAvailableSize returns the size that is available to be filled
 func (o *Order) GetAvailableSize() decimal.Decimal {
 	used := o.SizePending.Add(o.SizeFilled)

service/cancel_order.go

@@ -102,6 +102,9 @@ func (s *Service) CancelOrder(ctx context.Context, input CancelOrderInput) (*uui
 	})
 
 	logctx.Debug(ctx, "order cancelled", logger.String("orderId", order.Id.String()), logger.String("userId", order.UserId.String()), logger.String("size", order.Size.String()), logger.String("sizeFilled", order.SizeFilled.String()), logger.String("sizePending", order.SizePending.String()))
+
+	s.publishOrderEvent(ctx, order)
+
 	return &order.Id, nil
 }
 

service/cancel_orders_for_user.go

@@ -34,6 +34,9 @@ func (s *Service) CancelOrdersForUser(ctx context.Context, userId uuid.UUID, sym
 			if err != nil {
 				logctx.Error(ctx, "could not cancel order", logger.Error(err), logger.String("orderId", uid.String()))
 			}
+
+			s.publishOrderEvent(ctx, &order)
+
 			res = append(res, *uid)
 		}
 	}

service/create_order.go

@@ -79,5 +79,8 @@ func (s *Service) createNewOrder(ctx context.Context, input CreateOrderInput, us
 	}
 
 	logctx.Debug(ctx, "new order created", logger.String("ID", order.Id.String()), logger.String("price", order.Price.String()), logger.String("size", order.Size.String()))
+
+	s.publishOrderEvent(ctx, &order)
+
 	return order, nil
 }

service/evm_check_pending_txs_utils.go

@@ -73,6 +73,8 @@ func (e *EvmClient) ResolveSwap(ctx context.Context, swap models.Swap, isSuccess
 				continue
 			}
 
+			e.publishOrderEvent(ctx, &order)
+
 			if order.IsFilled() {
 				// add to filled orders if completely filled
 				filledOrders = append(filledOrders, order)

service/pub_sub.go

@@ -0,0 +1,58 @@
+package service
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+	"github.com/orbs-network/order-book/models"
+	"github.com/orbs-network/order-book/utils/logger"
+	"github.com/orbs-network/order-book/utils/logger/logctx"
+)
+
+func (s *Service) SubscribeUserOrders(ctx context.Context, userId uuid.UUID) (chan []byte, error) {
+	logctx.Debug(ctx, "subscribing to user orders", logger.String("userId", userId.String()))
+
+	eventKey := models.CreateUserOrdersEventKey(userId)
+
+	channel, err := s.orderBookStore.SubscribeToEvents(ctx, fmt.Sprintf("user_orders:%s", userId))
+	if err != nil {
+		logctx.Error(ctx, "failed to subscribe to user orders", logger.String("event", eventKey), logger.Error(err))
+		return nil, fmt.Errorf("failed to subscribe to user orders: %w", err)
+	}
+
+	return channel, nil
+}
+
+func (s *EvmClient) publishOrderEvent(ctx context.Context, order *models.Order) {
+	key, value, err := createOrderEvent(ctx, order)
+	if err != nil {
+		return
+	}
+
+	if err := s.orderBookStore.PublishEvent(ctx, key, value); err != nil {
+		logctx.Error(ctx, "failed to publish order event", logger.String("event", key), logger.Error(err))
+	}
+}
+
+func (s *Service) publishOrderEvent(ctx context.Context, order *models.Order) {
+	key, value, err := createOrderEvent(ctx, order)
+	if err != nil {
+		return
+	}
+
+	if err := s.orderBookStore.PublishEvent(ctx, key, value); err != nil {
+		logctx.Error(ctx, "failed to publish order event", logger.String("event", key), logger.Error(err))
+	}
+}
+
+func createOrderEvent(ctx context.Context, order *models.Order) (key string, value []byte, err error) {
+	value, err = order.ToJson()
+	if err != nil {
+		logctx.Error(ctx, "failed to marshal order to json", logger.Error(err))
+	}
+
+	key = models.CreateUserOrdersEventKey(order.UserId)
+
+	return key, value, err
+}

service/pub_sub_test.go

@@ -0,0 +1,36 @@
+package service_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/orbs-network/order-book/mocks"
+	"github.com/orbs-network/order-book/service"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestService_publishOrderEvent(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("should subscribe to user order event updates", func(t *testing.T) {
+		svc, _ := service.New(&mocks.MockOrderBookStore{
+			EventsChan: make(chan []byte),
+		}, &mocks.MockBcClient{})
+
+		channel, err := svc.SubscribeUserOrders(ctx, mocks.UserId)
+
+		assert.NotNil(t, channel)
+		assert.NoError(t, err)
+	})
+
+	t.Run("should return error when failed to subscribe to user order event updates", func(t *testing.T) {
+		svc, _ := service.New(&mocks.MockOrderBookStore{
+			Error: assert.AnError,
+		}, &mocks.MockBcClient{})
+
+		channel, err := svc.SubscribeUserOrders(ctx, mocks.UserId)
+
+		assert.Nil(t, channel)
+		assert.Error(t, err)
+	})
+}

service/service.go

@@ -22,6 +22,8 @@ type OrderBookService interface {
 	GetSymbols(ctx context.Context) ([]models.Symbol, error)
 	GetOpenOrdersForUser(ctx context.Context, userId uuid.UUID) (orders []models.Order, totalOrders int, err error)
 	GetFilledOrdersForUser(ctx context.Context, userId uuid.UUID) (orders []models.Order, totalOrders int, err error)
+	// Subscribe to order updates for a specific user
+	SubscribeUserOrders(ctx context.Context, userId uuid.UUID) (chan []byte, error)
 
 	// taker api - INSTEAD
 	GetQuote(ctx context.Context, symbol models.Symbol, side models.Side, inAmount decimal.Decimal, minOutAmount *decimal.Decimal) (models.QuoteRes, error)

service/taker.go

@@ -69,6 +69,7 @@ func (s *Service) BeginSwap(ctx context.Context, data models.QuoteRes) (models.B
 			logctx.Error(ctx, "Lock order Failed", logger.Error(err))
 			return models.BeginSwapRes{}, err
 		}
+		s.publishOrderEvent(ctx, &res.Orders[i])
 	}
 
 	// save
@@ -128,6 +129,7 @@ func (s *Service) AbortSwap(ctx context.Context, swapId uuid.UUID) error {
 				logctx.Error(ctx, "Unlock Failed", logger.Error(err))
 				return err
 			}
+			s.publishOrderEvent(ctx, order)
 			orders = append(orders, *order)
 		}
 		// cancelled orders
@@ -215,6 +217,7 @@ func (s *Service) FillSwap(ctx context.Context, swapId uuid.UUID) error {
 			} else {
 				openOrders = append(openOrders, *order)
 			}
+			s.publishOrderEvent(ctx, order)
 		}
 	}
 	// store partial orders