services/horizon/internal/actions: Add /order_book ingestion endpoint

exp/orderbook/graph.go

@@ -119,6 +119,55 @@ func (graph *OrderBookGraph) batch() *orderBookBatchedUpdates {
 	}
 }
 
+// findOffers returns all offers for a given trading pair
+// The offers will be sorted by price from cheapest to most expensive
+// The returned offers will span at most `maxPriceLevels` price levels
+func (graph *OrderBookGraph) findOffers(
+	selling, buying string, maxPriceLevels int,
+) []xdr.OfferEntry {
+	results := []xdr.OfferEntry{}
+	edges, ok := graph.edgesForSellingAsset[selling]
+	if !ok {
+		return results
+	}
+	offers, ok := edges[buying]
+	if !ok {
+		return results
+	}
+
+	for _, offer := range offers {
+		if len(results) == 0 || results[len(results)-1].Price != offer.Price {
+			maxPriceLevels--
+		}
+		if maxPriceLevels < 0 {
+			return results
+		}
+
+		results = append(results, offer)
+	}
+	return results
+}
+
+// FindAsksAndBids returns all asks and bids for a given trading pair
+// Asks consists of all offers which sell `selling` in exchange for `buying` sorted by
+// price (in terms of `buying`) from cheapest to most expensive
+// Bids consists of all offers which sell `buying` in exchange for `selling` sorted by
+// price (in terms of `selling`) from cheapest to most expensive
+// Both Asks and Bids will span at most `maxPriceLevels` price levels
+func (graph *OrderBookGraph) FindAsksAndBids(
+	selling, buying xdr.Asset, maxPriceLevels int,
+) ([]xdr.OfferEntry, []xdr.OfferEntry) {
+	buyingString := buying.String()
+	sellingString := selling.String()
+
+	graph.lock.RLock()
+	defer graph.lock.RUnlock()
+	asks := graph.findOffers(sellingString, buyingString, maxPriceLevels)
+	bids := graph.findOffers(buyingString, sellingString, maxPriceLevels)
+
+	return asks, bids
+}
+
 // add inserts a given offer into the order book graph
 func (graph *OrderBookGraph) add(offer xdr.OfferEntry) error {
 	if _, contains := graph.tradingPairForOffer[offer.OfferId]; contains {

exp/orderbook/graph_test.go

@@ -760,6 +760,143 @@ func TestRemoveOfferOrderBook(t *testing.T) {
 	}
 }
 
+func TestFindOffers(t *testing.T) {
+	graph := NewOrderBookGraph()
+
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		graph.findOffers(nativeAsset.String(), eurAsset.String(), 0),
+	)
+
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		graph.findOffers(nativeAsset.String(), eurAsset.String(), 5),
+	)
+
+	err := graph.
+		AddOffer(threeEurOffer).
+		AddOffer(eurOffer).
+		AddOffer(twoEurOffer).
+		Apply()
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		graph.findOffers(nativeAsset.String(), eurAsset.String(), 0),
+	)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{eurOffer, twoEurOffer},
+		graph.findOffers(nativeAsset.String(), eurAsset.String(), 2),
+	)
+
+	extraTwoEurOffers := []xdr.OfferEntry{}
+	for i := 0; i < 4; i++ {
+		otherTwoEurOffer := twoEurOffer
+		otherTwoEurOffer.OfferId += xdr.Int64(i + 17)
+		graph.AddOffer(otherTwoEurOffer)
+		extraTwoEurOffers = append(extraTwoEurOffers, otherTwoEurOffer)
+	}
+	if err := graph.Apply(); err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	assertOfferListEquals(
+		t,
+		append([]xdr.OfferEntry{eurOffer, twoEurOffer}, extraTwoEurOffers...),
+		graph.findOffers(nativeAsset.String(), eurAsset.String(), 2),
+	)
+	assertOfferListEquals(
+		t,
+		append(append([]xdr.OfferEntry{eurOffer, twoEurOffer}, extraTwoEurOffers...), threeEurOffer),
+		graph.findOffers(nativeAsset.String(), eurAsset.String(), 3),
+	)
+}
+
+func TestFindAsksAndBids(t *testing.T) {
+	graph := NewOrderBookGraph()
+
+	asks, bids := graph.FindAsksAndBids(nativeAsset, eurAsset, 0)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		asks,
+	)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		bids,
+	)
+
+	asks, bids = graph.FindAsksAndBids(nativeAsset, eurAsset, 5)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		asks,
+	)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		bids,
+	)
+
+	err := graph.
+		AddOffer(threeEurOffer).
+		AddOffer(eurOffer).
+		AddOffer(twoEurOffer).
+		Apply()
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	asks, bids = graph.FindAsksAndBids(nativeAsset, eurAsset, 0)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		asks,
+	)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{},
+		bids,
+	)
+
+	extraTwoEurOffers := []xdr.OfferEntry{}
+	for i := 0; i < 4; i++ {
+		otherTwoEurOffer := twoEurOffer
+		otherTwoEurOffer.OfferId += xdr.Int64(i + 17)
+		graph.AddOffer(otherTwoEurOffer)
+		extraTwoEurOffers = append(extraTwoEurOffers, otherTwoEurOffer)
+	}
+	if err := graph.Apply(); err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	sellEurOffer := twoEurOffer
+	sellEurOffer.Buying, sellEurOffer.Selling = sellEurOffer.Selling, sellEurOffer.Buying
+	sellEurOffer.OfferId = 35
+	if err := graph.AddOffer(sellEurOffer).Apply(); err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	asks, bids = graph.FindAsksAndBids(nativeAsset, eurAsset, 3)
+	assertOfferListEquals(
+		t,
+		append(append([]xdr.OfferEntry{eurOffer, twoEurOffer}, extraTwoEurOffers...), threeEurOffer),
+		asks,
+	)
+	assertOfferListEquals(
+		t,
+		[]xdr.OfferEntry{sellEurOffer},
+		bids,
+	)
+}
+
 func TestConsumeOffersForSellingAsset(t *testing.T) {
 	kp, err := keypair.Random()
 	if err != nil {

services/horizon/internal/actions/helpers_test.go

@@ -498,3 +498,26 @@ func testURLParams() map[string]string {
 		"long_12_asset_issuer": "GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
 	}
 }
+
+func makeRequest(
+	t *testing.T,
+	queryParams map[string]string,
+	routeParams map[string]string,
+) *http.Request {
+	request, err := http.NewRequest("GET", "/", nil)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+	query := url.Values{}
+	for key, value := range queryParams {
+		query.Set(key, value)
+	}
+	request.URL.RawQuery = query.Encode()
+
+	chiRouteContext := chi.NewRouteContext()
+	for key, value := range routeParams {
+		chiRouteContext.URLParams.Add(key, value)
+	}
+	ctx := context.WithValue(context.Background(), chi.RouteCtxKey, chiRouteContext)
+	return request.WithContext(ctx)
+}

services/horizon/internal/actions/offer_test.go

@@ -1,13 +1,9 @@
 package actions
 
 import (
-	"context"
-	"net/http"
-	"net/url"
 	"testing"
 	"time"
 
-	"github.com/go-chi/chi"
 	"github.com/stellar/go/protocols/horizon"
 	"github.com/stellar/go/services/horizon/internal/db2/core"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
@@ -63,50 +59,6 @@ var (
 	}
 )
 
-func makeOffersRequest(t *testing.T, queryParams map[string]string) *http.Request {
-	request, err := http.NewRequest("GET", "/", nil)
-	if err != nil {
-		t.Fatalf("unexpected error %v", err)
-	}
-	query := url.Values{}
-	for key, value := range queryParams {
-		query.Set(key, value)
-	}
-	request.URL.RawQuery = query.Encode()
-
-	ctx := context.WithValue(context.Background(), chi.RouteCtxKey, chi.NewRouteContext())
-	return request.WithContext(ctx)
-}
-
-func makeAccountOffersRequest(
-	t *testing.T,
-	accountID string,
-	queryParams map[string]string,
-) *http.Request {
-	request, err := http.NewRequest("GET", "/", nil)
-	if err != nil {
-		t.Fatalf("unexpected error %v", err)
-	}
-	query := url.Values{}
-	for key, value := range queryParams {
-		query.Set(key, value)
-	}
-	request.URL.RawQuery = query.Encode()
-
-	chiRouteContext := chi.NewRouteContext()
-	chiRouteContext.URLParams.Add("account_id", accountID)
-	ctx := context.WithValue(context.Background(), chi.RouteCtxKey, chiRouteContext)
-	return request.WithContext(ctx)
-}
-
-func pageableToOffers(t *testing.T, page []hal.Pageable) []horizon.Offer {
-	var offers []horizon.Offer
-	for _, entry := range page {
-		offers = append(offers, entry.(horizon.Offer))
-	}
-	return offers
-}
-
 func TestGetOffersHandler(t *testing.T) {
 	tt := test.Start(t)
 	defer tt.Finish()
@@ -133,7 +85,7 @@ func TestGetOffersHandler(t *testing.T) {
 	tt.Assert.NoError(q.UpsertOffer(usdOffer, 3))
 
 	t.Run("No filter", func(t *testing.T) {
-		records, err := handler.GetResourcePage(makeOffersRequest(t, map[string]string{}))
+		records, err := handler.GetResourcePage(makeRequest(t, map[string]string{}, map[string]string{}))
 		tt.Assert.NoError(err)
 		tt.Assert.Len(records, 3)
 
@@ -149,11 +101,12 @@ func TestGetOffersHandler(t *testing.T) {
 	})
 
 	t.Run("Filter by seller", func(t *testing.T) {
-		records, err := handler.GetResourcePage(makeOffersRequest(
+		records, err := handler.GetResourcePage(makeRequest(
 			t,
 			map[string]string{
 				"seller": issuer.Address(),
 			},
+			map[string]string{},
 		))
 		tt.Assert.NoError(err)
 		tt.Assert.Len(records, 2)
@@ -168,11 +121,12 @@ func TestGetOffersHandler(t *testing.T) {
 		asset := horizon.Asset{}
 		nativeAsset.Extract(&asset.Type, &asset.Code, &asset.Issuer)
 
-		records, err := handler.GetResourcePage(makeOffersRequest(
+		records, err := handler.GetResourcePage(makeRequest(
 			t,
 			map[string]string{
 				"selling_asset_type": asset.Type,
 			},
+			map[string]string{},
 		))
 		tt.Assert.NoError(err)
 		tt.Assert.Len(records, 2)
@@ -185,13 +139,14 @@ func TestGetOffersHandler(t *testing.T) {
 		asset = horizon.Asset{}
 		eurAsset.Extract(&asset.Type, &asset.Code, &asset.Issuer)
 
-		records, err = handler.GetResourcePage(makeOffersRequest(
+		records, err = handler.GetResourcePage(makeRequest(
 			t,
 			map[string]string{
 				"selling_asset_type":   asset.Type,
 				"selling_asset_code":   asset.Code,
 				"selling_asset_issuer": asset.Issuer,
 			},
+			map[string]string{},
 		))
 		tt.Assert.NoError(err)
 		tt.Assert.Len(records, 1)
@@ -204,13 +159,14 @@ func TestGetOffersHandler(t *testing.T) {
 		asset := horizon.Asset{}
 		eurAsset.Extract(&asset.Type, &asset.Code, &asset.Issuer)
 
-		records, err := handler.GetResourcePage(makeOffersRequest(
+		records, err := handler.GetResourcePage(makeRequest(
 			t,
 			map[string]string{
 				"buying_asset_type":   asset.Type,
 				"buying_asset_code":   asset.Code,
 				"buying_asset_issuer": asset.Issuer,
 			},
+			map[string]string{},
 		))
 		tt.Assert.NoError(err)
 		tt.Assert.Len(records, 2)
@@ -223,13 +179,14 @@ func TestGetOffersHandler(t *testing.T) {
 		asset = horizon.Asset{}
 		usdAsset.Extract(&asset.Type, &asset.Code, &asset.Issuer)
 
-		records, err = handler.GetResourcePage(makeOffersRequest(
+		records, err = handler.GetResourcePage(makeRequest(
 			t,
 			map[string]string{
 				"buying_asset_type":   asset.Type,
 				"buying_asset_code":   asset.Code,
 				"buying_asset_issuer": asset.Issuer,
 			},
+			map[string]string{},
 		))
 		tt.Assert.NoError(err)
 		tt.Assert.Len(records, 1)
@@ -256,7 +213,7 @@ func TestGetAccountOffersHandler(t *testing.T) {
 	tt.Assert.NoError(q.UpsertOffer(usdOffer, 3))
 
 	records, err := handler.GetResourcePage(
-		makeAccountOffersRequest(t, issuer.Address(), map[string]string{}),
+		makeRequest(t, map[string]string{}, map[string]string{"account_id": issuer.Address()}),
 	)
 	tt.Assert.NoError(err)
 	tt.Assert.Len(records, 2)
@@ -267,3 +224,11 @@ func TestGetAccountOffersHandler(t *testing.T) {
 		tt.Assert.Equal(issuer.Address(), offer.Seller)
 	}
 }
+
+func pageableToOffers(t *testing.T, page []hal.Pageable) []horizon.Offer {
+	var offers []horizon.Offer
+	for _, entry := range page {
+		offers = append(offers, entry.(horizon.Offer))
+	}
+	return offers
+}