services/horizon: Add /liquidity_pools endpoint

services/horizon/internal/actions/liquidity_pool.go

@@ -1,13 +1,19 @@
 package actions
 
 import (
+	"context"
 	"net/http"
+	"strings"
 
 	protocol "github.com/stellar/go/protocols/horizon"
 	horizonContext "github.com/stellar/go/services/horizon/internal/context"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
+	"github.com/stellar/go/services/horizon/internal/ledger"
 	"github.com/stellar/go/services/horizon/internal/resourceadapter"
 	"github.com/stellar/go/support/errors"
+	"github.com/stellar/go/support/render/hal"
+	"github.com/stellar/go/support/render/problem"
+	"github.com/stellar/go/xdr"
 )
 
 // GetLiquidityPoolByIDHandler is the action handler for all end-points returning a liquidity pool.
@@ -51,3 +57,107 @@ func (handler GetLiquidityPoolByIDHandler) GetResource(w HeaderWriter, r *http.R
 
 	return resource, nil
 }
+
+// LiquidityPoolsQuery query struct for liquidity_pools end-point
+type LiquidityPoolsQuery struct {
+	Reserves string `schema:"reserves" valid:"optional"`
+
+	reserves []xdr.Asset
+}
+
+// Validate validates and parses the query
+func (q *LiquidityPoolsQuery) Validate() error {
+	assets := []xdr.Asset{}
+	reserves := strings.Split(q.Reserves, ",")
+	reservesErr := problem.MakeInvalidFieldProblem(
+		"reserves",
+		errors.New("Invalid reserves, should be comma-separated list of assets in canonical form"),
+	)
+	for _, reserve := range reserves {
+		if reserve == "" {
+			continue
+		}
+		switch reserve {
+		case "native":
+			assets = append(assets, xdr.MustNewNativeAsset())
+		default:
+			parts := strings.Split(reserve, ":")
+			if len(parts) != 2 {
+				return reservesErr
+			}
+			asset, err := xdr.NewCreditAsset(parts[0], parts[1])
+			if err != nil {
+				return reservesErr
+			}
+			assets = append(assets, asset)
+		}
+	}
+	q.reserves = assets
+	return nil
+}
+
+type GetLiquidityPoolsHandler struct {
+	LedgerState *ledger.State
+}
+
+// GetResourcePage returns a page of claimable balances.
+func (handler GetLiquidityPoolsHandler) GetResourcePage(w HeaderWriter, r *http.Request) ([]hal.Pageable, error) {
+	ctx := r.Context()
+	qp := LiquidityPoolsQuery{}
+	err := getParams(&qp, r)
+	if err != nil {
+		return nil, err
+	}
+
+	pq, err := GetPageQuery(handler.LedgerState, r, DisableCursorValidation)
+	if err != nil {
+		return nil, err
+	}
+
+	query := history.LiquidityPoolsQuery{
+		PageQuery: pq,
+		Assets:    qp.reserves,
+	}
+
+	historyQ, err := horizonContext.HistoryQFromRequest(r)
+	if err != nil {
+		return nil, err
+	}
+
+	liquidityPools, err := handler.getLiquidityPoolsPage(ctx, historyQ, query)
+	if err != nil {
+		return nil, err
+	}
+
+	return liquidityPools, nil
+}
+
+func (handler GetLiquidityPoolsHandler) getLiquidityPoolsPage(ctx context.Context, historyQ *history.Q, query history.LiquidityPoolsQuery) ([]hal.Pageable, error) {
+	records, err := historyQ.GetLiquidityPools(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	ledgerCache := history.LedgerCache{}
+	for _, record := range records {
+		ledgerCache.Queue(int32(record.LastModifiedLedger))
+	}
+	if err := ledgerCache.Load(ctx, historyQ); err != nil {
+		return nil, errors.Wrap(err, "failed to load ledger batch")
+	}
+
+	var liquidityPools []hal.Pageable
+	for _, record := range records {
+		var response protocol.LiquidityPool
+
+		var ledger *history.Ledger
+		if l, ok := ledgerCache.Records[int32(record.LastModifiedLedger)]; ok {
+			ledger = &l
+		}
+
+		resourceadapter.PopulateLiquidityPool(ctx, &response, record, ledger)
+		liquidityPools = append(liquidityPools, response)
+	}
+
+	return liquidityPools, nil
+}

services/horizon/internal/actions/liquidity_pool_test.go

@@ -88,3 +88,108 @@ func TestGetLiquidityPoolByID(t *testing.T) {
 	tt.Assert.Equal("id", p.Extras["invalid_field"])
 	tt.Assert.Equal("0000001112122 does not validate as sha256", p.Extras["reason"])
 }
+
+func TestGetLiquidityPools(t *testing.T) {
+	tt := test.Start(t)
+	defer tt.Finish()
+	test.ResetHorizonDB(t, tt.HorizonDB)
+	q := &history.Q{tt.HorizonSession()}
+
+	builder := q.NewLiquidityPoolsBatchInsertBuilder(2)
+	err := builder.Add(tt.Ctx, history.LiquidityPool{
+		PoolID:         "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
+		Type:           xdr.LiquidityPoolTypeLiquidityPoolConstantProduct,
+		Fee:            30,
+		TrustlineCount: 100,
+		ShareCount:     200,
+		AssetReserves: history.LiquidityPoolAssetReserves{
+			{
+				Asset:   xdr.MustNewNativeAsset(),
+				Reserve: 100,
+			},
+			{
+				Asset:   xdr.MustNewCreditAsset("USD", "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
+				Reserve: 200,
+			},
+		},
+		LastModifiedLedger: 100,
+	})
+	tt.Assert.NoError(err)
+	err = builder.Add(tt.Ctx, history.LiquidityPool{
+		PoolID:         "d827bf10a721d217de3cd9ab3f10198a54de558c093a511ec426028618df2633",
+		Type:           xdr.LiquidityPoolTypeLiquidityPoolConstantProduct,
+		Fee:            30,
+		TrustlineCount: 300,
+		ShareCount:     400,
+		AssetReserves: history.LiquidityPoolAssetReserves{
+			{
+				Asset:   xdr.MustNewCreditAsset("EUR", "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
+				Reserve: 300,
+			},
+			{
+				Asset:   xdr.MustNewCreditAsset("USD", "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
+				Reserve: 400,
+			},
+		},
+		LastModifiedLedger: 100,
+	})
+	tt.Assert.NoError(err)
+	err = builder.Exec(tt.Ctx)
+	tt.Assert.NoError(err)
+
+	handler := GetLiquidityPoolsHandler{}
+	response, err := handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
+		t,
+		map[string]string{},
+		map[string]string{},
+		q,
+	))
+	tt.Assert.NoError(err)
+	tt.Assert.Len(response, 2)
+
+	resource := response[0].(protocol.LiquidityPool)
+	tt.Assert.Equal("ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad", resource.ID)
+	tt.Assert.Equal("constant_product", resource.Type)
+	tt.Assert.Equal(uint32(30), resource.FeeBP)
+	tt.Assert.Equal(uint64(100), resource.TotalTrustlines)
+	tt.Assert.Equal(uint64(200), resource.TotalShares)
+
+	tt.Assert.Equal("native", resource.Reserves[0].Asset)
+	tt.Assert.Equal("0.0000100", resource.Reserves[0].Amount)
+
+	tt.Assert.Equal("USD:GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML", resource.Reserves[1].Asset)
+	tt.Assert.Equal("0.0000200", resource.Reserves[1].Amount)
+
+	resource = response[1].(protocol.LiquidityPool)
+	tt.Assert.Equal("d827bf10a721d217de3cd9ab3f10198a54de558c093a511ec426028618df2633", resource.ID)
+	tt.Assert.Equal("constant_product", resource.Type)
+	tt.Assert.Equal(uint32(30), resource.FeeBP)
+	tt.Assert.Equal(uint64(300), resource.TotalTrustlines)
+	tt.Assert.Equal(uint64(400), resource.TotalShares)
+
+	tt.Assert.Equal("EUR:GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML", resource.Reserves[0].Asset)
+	tt.Assert.Equal("0.0000300", resource.Reserves[0].Amount)
+
+	tt.Assert.Equal("USD:GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML", resource.Reserves[1].Asset)
+	tt.Assert.Equal("0.0000400", resource.Reserves[1].Amount)
+
+	response, err = handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
+		t,
+		map[string]string{"reserves": "native"},
+		map[string]string{},
+		q,
+	))
+	tt.Assert.NoError(err)
+	tt.Assert.Len(response, 1)
+
+	response, err = handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
+		t,
+		map[string]string{"cursor": "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"},
+		map[string]string{},
+		q,
+	))
+	tt.Assert.NoError(err)
+	tt.Assert.Len(response, 1)
+	resource = response[0].(protocol.LiquidityPool)
+	tt.Assert.Equal("d827bf10a721d217de3cd9ab3f10198a54de558c093a511ec426028618df2633", resource.ID)
+}

services/horizon/internal/httpx/router.go

@@ -165,6 +165,7 @@ func (r *Router) addRoutes(config *RouterConfig, rateLimiter *throttled.HTTPRate
 		})
 
 		r.Route("/liquidity_pools", func(r chi.Router) {
+			r.With(stateMiddleware.Wrap).Method(http.MethodGet, "/", restPageHandler(ledgerState, actions.GetLiquidityPoolsHandler{LedgerState: ledgerState}))
 			r.With(stateMiddleware.Wrap).Method(http.MethodGet, "/{id}", ObjectActionHandler{actions.GetLiquidityPoolByIDHandler{}})
 		})
 

services/horizon/internal/integration/protocol18_test.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/stellar/go/ingest"
+	"github.com/stellar/go/clients/horizonclient"
 	"github.com/stellar/go/services/horizon/internal/test/integration"
 	"github.com/stellar/go/txnbuild"
 	"github.com/stellar/go/xdr"
@@ -48,7 +48,7 @@ func TestCreateLiquidityPool(t *testing.T) {
 	keys, accounts := itest.CreateAccounts(1, "1000")
 	shareKeys, shareAccount := keys[0], accounts[0]
 
-	resp := itest.MustSubmitOperations(shareAccount, shareKeys,
+	itest.MustSubmitOperations(shareAccount, shareKeys,
 		&txnbuild.ChangeTrust{
 			Line: txnbuild.ChangeTrustAssetWrapper{
 				Asset: txnbuild.CreditAsset{
@@ -73,28 +73,16 @@ func TestCreateLiquidityPool(t *testing.T) {
 		},
 	)
 
-	// TODO rewrite it to use /liquidity_pools when ready
+	pools, err := itest.Client().LiquidityPools(horizonclient.LiquidityPoolsRequest{})
+	tt.NoError(err)
+	tt.Len(pools.Embedded.Records, 1)
+
 	expectedID, err := xdr.NewPoolId(
 		xdr.MustNewNativeAsset(),
 		xdr.MustNewCreditAsset("USD", master.Address()),
 		30,
 	)
 	tt.NoError(err)
 
-	var transactionMeta xdr.TransactionMeta
-	err = xdr.SafeUnmarshalBase64(resp.ResultMetaXdr, &transactionMeta)
-	tt.NoError(err)
-	changes := ingest.GetChangesFromLedgerEntryChanges(transactionMeta.OperationsMeta()[1].Changes)
-	found := false
-	for _, change := range changes {
-		if change.Type != xdr.LedgerEntryTypeLiquidityPool {
-			continue
-		}
-
-		tt.Nil(change.Pre)
-		tt.NotNil(change.Post)
-		tt.Equal(expectedID, change.Post.Data.MustLiquidityPool().LiquidityPoolId)
-		found = true
-	}
-	tt.True(found, "liquidity pool not found in meta")
+	tt.Equal(xdr.Hash(expectedID).HexString(), pools.Embedded.Records[0].ID)
 }

services/horizon/internal/resourceadapter/liquidity_pool.go

@@ -45,7 +45,7 @@ func PopulateLiquidityPool(
 	lb := hal.LinkBuilder{Base: horizonContext.BaseURL(ctx)}
 	self := fmt.Sprintf("/liquidity_pools/%s", dest.ID)
 	dest.Links.Self = lb.Link(self)
-	dest.PT = fmt.Sprintf("%d-%s", liquidityPool.LastModifiedLedger, dest.ID)
+	dest.PT = dest.ID
 	dest.Links.Transactions = lb.PagedLink(self, "transactions")
 	dest.Links.Operations = lb.PagedLink(self, "operations")
 	return nil