SQS-412 | Active Orders Query: SSE

Implements SSE for Active Orders Query endpoint.

app/sidecar_query_server.go

@@ -203,7 +203,7 @@ func NewSideCarQueryServer(appCodec codec.Codec, config domain.Config, logger lo
 
 	// HTTP handlers
 	poolsHttpDelivery.NewPoolsHandler(e, poolsUseCase)
-	passthroughHttpDelivery.NewPassthroughHandler(e, passthroughUseCase, orderBookUseCase)
+	passthroughHttpDelivery.NewPassthroughHandler(e, passthroughUseCase, orderBookUseCase, logger)
 	systemhttpdelivery.NewSystemHandler(e, config, logger, chainInfoUseCase)
 	if err := tokenshttpdelivery.NewTokensHandler(e, *config.Pricing, tokensUseCase, pricingSimpleRouterUsecase, logger); err != nil {
 		return nil, err

delivery/http/event.go

@@ -0,0 +1,97 @@
+package http
+
+import (
+	"encoding/json"
+	"bytes"
+	"fmt"
+	"io"
+
+	"github.com/labstack/echo/v4"
+)
+
+// Event represents Server-Sent Event.
+// SSE explanation: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
+type Event struct {
+	// ID is used to set the EventSource object's last event ID value.
+	ID []byte
+	// Data field is for the message. When the EventSource receives multiple consecutive lines
+	// that begin with data:, it concatenates them, inserting a newline character between each one.
+	// Trailing newlines are removed.
+	Data []byte
+	// Event is a string identifying the type of event described. If this is specified, an event
+	// will be dispatched on the browser to the listener for the specified event name; the website
+	// source code should use addEventListener() to listen for named events. The onmessage handler
+	// is called if no event name is specified for a message.
+	Event []byte
+	// Retry is the reconnection time. If the connection to the server is lost, the browser will
+	// wait for the specified time before attempting to reconnect. This must be an integer, specifying
+	// the reconnection time in milliseconds. If a non-integer value is specified, the field is ignored.
+	Retry []byte
+	// Comment line can be used to prevent connections from timing out; a server can send a comment
+	// periodically to keep the connection alive.
+	Comment []byte
+}
+
+// MarshalTo marshals Event to given Writer
+func (ev *Event) MarshalTo(w io.Writer) error {
+	// Marshalling part is taken from: https://github.com/r3labs/sse/blob/c6d5381ee3ca63828b321c16baa008fd6c0b4564/http.go#L16
+	if len(ev.Data) == 0 && len(ev.Comment) == 0 {
+		return nil
+	}
+
+	if len(ev.Data) > 0 {
+		if _, err := fmt.Fprintf(w, "id: %s\n", ev.ID); err != nil {
+			return err
+		}
+
+		sd := bytes.Split(ev.Data, []byte("\n"))
+		for i := range sd {
+			if _, err := fmt.Fprintf(w, "data: %s\n", sd[i]); err != nil {
+				return err
+			}
+		}
+
+		if len(ev.Event) > 0 {
+			if _, err := fmt.Fprintf(w, "event: %s\n", ev.Event); err != nil {
+				return err
+			}
+		}
+
+		if len(ev.Retry) > 0 {
+			if _, err := fmt.Fprintf(w, "retry: %s\n", ev.Retry); err != nil {
+				return err
+			}
+		}
+	}
+
+	if len(ev.Comment) > 0 {
+		if _, err := fmt.Fprintf(w, ": %s\n", ev.Comment); err != nil {
+			return err
+		}
+	}
+
+	if _, err := fmt.Fprint(w, "\n"); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func WriteEvent(w *echo.Response, data any) error {
+	b, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+
+	event := Event{
+		Data: b,
+	}
+
+	if err := event.MarshalTo(w); err != nil {
+		return err
+	}
+
+	w.Flush()
+
+	return nil
+}

domain/mocks/orderbook_usecase_mock.go

@@ -12,9 +12,10 @@ var _ mvc.OrderBookUsecase = &OrderbookUsecaseMock{}
 
 // OrderbookUsecaseMock is a mock implementation of the RouterUsecase interface
 type OrderbookUsecaseMock struct {
-	ProcessPoolFunc     func(ctx context.Context, pool sqsdomain.PoolI) error
-	GetAllTicksFunc     func(poolID uint64) (map[int64]orderbookdomain.OrderbookTick, bool)
-	GetActiveOrdersFunc func(ctx context.Context, address string) ([]orderbookdomain.LimitOrder, bool, error)
+	ProcessPoolFunc           func(ctx context.Context, pool sqsdomain.PoolI) error
+	GetAllTicksFunc           func(poolID uint64) (map[int64]orderbookdomain.OrderbookTick, bool)
+	GetActiveOrdersFunc       func(ctx context.Context, address string) ([]orderbookdomain.LimitOrder, bool, error)
+	GetActiveOrdersStreamFunc func(ctx context.Context, address string) <-chan orderbookdomain.ActiveOrder
 }
 
 func (m *OrderbookUsecaseMock) ProcessPool(ctx context.Context, pool sqsdomain.PoolI) error {
@@ -37,3 +38,10 @@ func (m *OrderbookUsecaseMock) GetActiveOrders(ctx context.Context, address stri
 	}
 	panic("unimplemented")
 }
+
+func (m *OrderbookUsecaseMock) GetActiveOrdersStream(ctx context.Context, address string) <-chan orderbookdomain.ActiveOrder {
+	if m.GetActiveOrdersStreamFunc != nil {
+		return m.GetActiveOrdersStreamFunc(ctx, address)
+	}
+	panic("unimplemented")
+}

domain/mvc/orderbook.go

@@ -16,4 +16,7 @@ type OrderBookUsecase interface {
 
 	// GetOrder returns all active orderbook orders for a given address.
 	GetActiveOrders(ctx context.Context, address string) ([]orderbookdomain.LimitOrder, bool, error)
+
+	// GetActiveOrdersStream returns a channel for feching active orderbook orders for a given address.
+	GetActiveOrdersStream(ctx context.Context, address string)  <-chan orderbookdomain.ActiveOrder
 }

domain/orderbook/order.go

@@ -93,3 +93,10 @@ type LimitOrder struct {
 	BaseAsset        Asset        `json:"base_asset"`
 	PlacedTx         *string      `json:"placed_tx,omitempty"`
 }
+
+
+type ActiveOrder struct {
+	LimitOrders  []LimitOrder // The channel on which the orders are delivered.
+	IsBestEffort bool
+	Error        error
+}

orderbook/usecase/orderbook_usecase.go

@@ -137,6 +137,63 @@ func (o *OrderbookUseCaseImpl) ProcessPool(ctx context.Context, pool sqsdomain.P
 	return nil
 }
 
+// GetActiveOrdersStream implements mvc.OrderBookUsecase.
+func (o *OrderbookUseCaseImpl) GetActiveOrdersStream(ctx context.Context, address string)  <-chan orderbookdomain.ActiveOrder {
+	// Fetch orders duration
+	d := 5 * time.Second
+
+	// Result channel
+	c := make(chan orderbookdomain.ActiveOrder, 50) // buffered channel to avoid blocking
+
+	// Function to fetch orders
+	fetchOrders := func(ctx context.Context) {
+		orderbooks, err := o.poolsUsecease.GetAllCanonicalOrderbookPoolIDs()
+		if err != nil {
+			c <- orderbookdomain.ActiveOrder{
+				Error: types.FailedGetAllCanonicalOrderbookPoolIDsError{Err: err},
+			}
+		}
+
+		for _, orderbook := range orderbooks {
+			go func(orderbook domain.CanonicalOrderBooksResult) {
+				limitOrders, isBestEffort, err := o.processOrderBookActiveOrders(ctx, orderbook, address)
+				if len(limitOrders) == 0 && err == nil {
+					return // skip empty orders
+				}
+
+				select {
+				case c <- orderbookdomain.ActiveOrder{
+					IsBestEffort: isBestEffort,
+					LimitOrders:  limitOrders,
+					Error:        err,
+				}:
+				case <-ctx.Done():
+					return
+				}
+			}(orderbook)
+		}
+	}
+
+	// Fetch orders immediately on start
+	go fetchOrders(ctx)
+
+	// Pull orders periodically based on duration
+	go func() {
+		ticker := time.NewTicker(d)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				fetchOrders(ctx)
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return c
+}
+
 // GetActiveOrders implements mvc.OrderBookUsecase.
 func (o *OrderbookUseCaseImpl) GetActiveOrders(ctx context.Context, address string) ([]orderbookdomain.LimitOrder, bool, error) {
 	orderbooks, err := o.poolsUsecease.GetAllCanonicalOrderbookPoolIDs()

passthrough/delivery/http/passthrough_handler.go

@@ -7,16 +7,20 @@ import (
 	"github.com/osmosis-labs/sqs/domain"
 	"github.com/osmosis-labs/sqs/domain/mvc"
 	_ "github.com/osmosis-labs/sqs/domain/passthrough"
+	"github.com/osmosis-labs/sqs/log"
 	"github.com/osmosis-labs/sqs/orderbook/types"
 
 	"github.com/labstack/echo/v4"
 	"go.opentelemetry.io/otel/trace"
+
+	"go.uber.org/zap"
 )
 
 // PassthroughHandler is the http handler for passthrough use case
 type PassthroughHandler struct {
 	PUsecase mvc.PassthroughUsecase
 	OUsecase mvc.OrderBookUsecase
+	Logger   log.Logger
 }
 
 const resourcePrefix = "/passthrough"
@@ -26,14 +30,21 @@ func formatPassthroughResource(resource string) string {
 }
 
 // NewPassthroughHandler will initialize the pools/ resources endpoint
-func NewPassthroughHandler(e *echo.Echo, ptu mvc.PassthroughUsecase, ou mvc.OrderBookUsecase) {
+func NewPassthroughHandler(e *echo.Echo, ptu mvc.PassthroughUsecase, ou mvc.OrderBookUsecase, logger log.Logger) {
 	handler := &PassthroughHandler{
 		PUsecase: ptu,
 		OUsecase: ou,
+		Logger:   logger,
 	}
 
 	e.GET(formatPassthroughResource("/portfolio-assets/:address"), handler.GetPortfolioAssetsByAddress)
 	e.GET(formatPassthroughResource("/active-orders"), handler.GetActiveOrders)
+	e.GET(formatPassthroughResource("/active-orders"), func(c echo.Context) error {
+		if c.QueryParam("sse") != "" {
+			return handler.GetActiveOrdersStream(c) // Server-Sent Events (SSE)
+		}
+		return handler.GetActiveOrders(c)
+	})
 }
 
 // @Summary Returns portfolio assets associated with the given address by category.
@@ -61,6 +72,58 @@ func (a *PassthroughHandler) GetPortfolioAssetsByAddress(c echo.Context) error {
 	return c.JSON(http.StatusOK, portfolioAssetsResult)
 }
 
+func (a *PassthroughHandler) GetActiveOrdersStream(c echo.Context) (err error) {
+	ctx := c.Request().Context()
+
+	span := trace.SpanFromContext(ctx)
+	defer func() {
+		if err != nil {
+			span.RecordError(err)
+			// nolint:errcheck // ignore error
+			c.JSON(domain.GetStatusCode(err), domain.ResponseError{Message: err.Error()})
+		}
+
+		// Note: we do not end the span here as it is ended in the middleware.
+	}()
+
+	var req types.GetActiveOrdersRequest
+	if err := deliveryhttp.UnmarshalRequest(c, &req); err != nil {
+		return c.JSON(http.StatusBadRequest, domain.ResponseError{Message: err.Error()})
+	}
+
+	// Validate the request
+	if err := req.Validate(); err != nil {
+		return c.JSON(http.StatusBadRequest, domain.ResponseError{Message: err.Error()})
+	}
+
+	w := c.Response()
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	ch := a.OUsecase.GetActiveOrdersStream(ctx, req.UserOsmoAddress)
+
+	for {
+		select {
+		case <-c.Request().Context().Done():
+			return c.NoContent(http.StatusOK)
+		case orders, ok := <-ch:
+			if !ok {
+				return c.NoContent(http.StatusOK)
+			}
+
+			if orders.Error != nil {
+				a.Logger.Error("GET "+c.Request().URL.String(), zap.Error(orders.Error))
+			}
+
+			deliveryhttp.WriteEvent(
+				w,
+				types.NewGetAllOrderResponse(orders.LimitOrders, orders.IsBestEffort),
+			)
+		}
+	}
+}
+
 // @Summary Returns all active orderbook orders associated with the given address.
 // @Description The returned data represents all active orders for all orderbooks available for the specified address.
 //