Offers schema 9 (stellar#972)

Updates offer table queries to use new Core DB schema 9
(stellar/stellar-core#1957). It also fixes SignersByAddress to support
NULL values (close stellar#967).

services/horizon/internal/actions_offer_test.go

@@ -24,6 +24,16 @@ func TestOfferActions_Index(t *testing.T) {
  //test last modified timestamp
  var records []map[string]interface{}
  ht.UnmarshalPage(w.Body, &records)
+
+ // Test asset fields population
+ ht.Assert.Equal("credit_alphanum4", records[2]["selling"].(map[string]interface{})["asset_type"])
+ ht.Assert.Equal("EUR", records[2]["selling"].(map[string]interface{})["asset_code"])
+ ht.Assert.Equal("GCQPYGH4K57XBDENKKX55KDTWOTK5WDWRQOH2LHEDX3EKVIQRLMESGBG", records[2]["selling"].(map[string]interface{})["asset_issuer"])
+
+ ht.Assert.Equal("credit_alphanum4", records[2]["buying"].(map[string]interface{})["asset_type"])
+ ht.Assert.Equal("USD", records[2]["buying"].(map[string]interface{})["asset_code"])
+ ht.Assert.Equal("GC23QF2HUE52AMXUFUH3AYJAXXGXXV2VHXYYR6EYXETPKDXZSAW67XO4", records[2]["buying"].(map[string]interface{})["asset_issuer"])
+
  t2018, err := time.Parse("2006-01-02", "2018-01-01")
  ht.Assert.NoError(err)
  recordTime, err := time.Parse("2006-01-02T15:04:05Z", records[2]["last_modified_time"].(string))

services/horizon/internal/db2/core/main.go

@@ -49,13 +49,8 @@ type Offer struct {
  SellerID string `db:"sellerid"`
  OfferID int64 `db:"offerid"`
 
- SellingAssetType xdr.AssetType `db:"sellingassettype"`
- SellingAssetCode null.String `db:"sellingassetcode"`
- SellingIssuer null.String `db:"sellingissuer"`
-
- BuyingAssetType xdr.AssetType `db:"buyingassettype"`
- BuyingAssetCode null.String `db:"buyingassetcode"`
- BuyingIssuer null.String `db:"buyingissuer"`
+ SellingAsset xdr.Asset `db:"sellingasset"`
+ BuyingAsset xdr.Asset `db:"buyingasset"`
 
  Amount xdr.Int64 `db:"amount"`
  Pricen int32 `db:"pricen"`
@@ -65,6 +60,26 @@ type Offer struct {
  Lastmodified int32 `db:"lastmodified"`
 }
 
+// get returns Offer. Useful in `internalOffer` context, when Offer is embedded.
+func (o Offer) get() Offer {
+ return o
+}
+
+// internalOffer is row of data from the `offers` table from stellar-core used
+// internally only to support schema <=8.
+type internalOffer struct {
+ Offer
+
+ // Schema v8 fields, for compatibility only.
+ SellingAssetType xdr.AssetType `db:"sellingassettype"`
+ SellingAssetCode null.String `db:"sellingassetcode"`
+ SellingIssuer null.String `db:"sellingissuer"`
+
+ BuyingAssetType xdr.AssetType `db:"buyingassettype"`
+ BuyingAssetCode null.String `db:"buyingassetcode"`
+ BuyingIssuer null.String `db:"buyingissuer"`
+}
+
 // OrderBookSummaryPriceLevel is a collapsed view of multiple offers at the same price that
 // contains the summed amount from all the member offers. Used by OrderBookSummary
 type OrderBookSummaryPriceLevel struct {

services/horizon/internal/db2/core/main_test.go

@@ -3,7 +3,9 @@ package core
 import (
  "testing"
 
+ "github.com/stellar/go/services/horizon/internal/db2"
  "github.com/stellar/go/services/horizon/internal/test"
+ "github.com/stellar/go/xdr"
 )
 
 func TestLatestLedger(t *testing.T) {
@@ -99,6 +101,62 @@ func TestSchemaVersion8(t *testing.T) {
  tt.Assert.Equal("GAFEES4MDE5Z7Q6JBB2BYMLS7YWEHTPNR7ICANZA7TAOLMSRELE4H4S2", signers[0].Publickey)
  tt.Assert.Equal(int32(2), signers[0].Weight)
  }
+
+ pq, err := db2.NewPageQuery("", true, "asc", db2.DefaultPageSize)
+ if !tt.Assert.NoError(err) {
+ return
+ }
+
+ var offers []Offer
+ err = q.OffersByAddress(&offers, "GAXMF43TGZHW3QN3REOUA2U5PW5BTARXGGYJ3JIFHW3YT6QRKRL3CPPU", pq)
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(1, len(offers))
+ tt.Assert.True(offers[0].SellingAsset.Equals(xdr.MustNewNativeAsset()))
+ tt.Assert.True(offers[0].BuyingAsset.Equals(xdr.MustNewCreditAsset("USD", "GAXMF43TGZHW3QN3REOUA2U5PW5BTARXGGYJ3JIFHW3YT6QRKRL3CPPU")))
+ }
+
+ offers = []Offer{}
+ err = q.OffersByAddress(&offers, "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD", pq)
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(4, len(offers))
+
+ tt.Assert.True(offers[0].SellingAsset.Equals(xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD")))
+ tt.Assert.True(offers[0].BuyingAsset.Equals(xdr.MustNewNativeAsset()))
+ }
+
+ var assets []xdr.Asset
+ err = q.ConnectedAssets(&assets, xdr.MustNewNativeAsset())
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(2, len(assets))
+ connectedAsset := xdr.MustNewCreditAsset("USD", "GAXMF43TGZHW3QN3REOUA2U5PW5BTARXGGYJ3JIFHW3YT6QRKRL3CPPU")
+ tt.Assert.True(assets[0].Equals(connectedAsset))
+ }
+
+ assets = []xdr.Asset{}
+ err = q.ConnectedAssets(&assets, xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD"))
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(1, len(assets))
+ connectedAsset := xdr.MustNewNativeAsset()
+ tt.Assert.True(assets[0].Equals(connectedAsset))
+ }
+
+ var orderbookSummary OrderBookSummary
+ err = q.GetOrderBookSummary(&orderbookSummary, xdr.MustNewNativeAsset(), xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD"), 10)
+ if tt.Assert.NoError(err) {
+ checkOrderBookRow(tt, orderbookSummary[0], "ask", int32(1), int32(1), float64(1), 300000000)
+ checkOrderBookRow(tt, orderbookSummary[1], "ask", int32(3), int32(1), float64(3), 400000000)
+
+ checkOrderBookRow(tt, orderbookSummary[2], "bid", int32(1), int32(2), float64(0.5), 200000000)
+ checkOrderBookRow(tt, orderbookSummary[3], "bid", int32(1), int32(1), float64(1), 100000000)
+ }
+}
+
+func checkOrderBookRow(tt *test.T, row OrderBookSummaryPriceLevel, typ string, pricen, priced int32, pricef float64, amount int64) {
+ tt.Assert.Equal(typ, row.Type)
+ tt.Assert.Equal(pricen, row.Pricen)
+ tt.Assert.Equal(priced, row.Priced)
+ tt.Assert.Equal(pricef, row.Pricef)
+ tt.Assert.Equal(amount, row.Amount)
 }
 
 func TestSchemaVersion9(t *testing.T) {
@@ -138,4 +196,77 @@ func TestSchemaVersion9(t *testing.T) {
  tt.Assert.Equal("GC7BWB2ME4LII3TVWTHUIT7KGJXU4D5M6JUNLQ57WA7JERDNSAEXLOAN", signers[0].Publickey)
  tt.Assert.Equal(int32(10), signers[0].Weight)
  }
+
+ var signers2 []Signer
+ err = q.SignersByAddress(&signers2, "GD7HOGYRECGFKFR2GGOWEF2FT3DVR3GU4K7BVRGGPWVSXAVKGSYKTXOH")
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(0, len(signers2))
+ }
+
+ pq, err := db2.NewPageQuery("", true, "asc", db2.DefaultPageSize)
+ if !tt.Assert.NoError(err) {
+ return
+ }
+
+ var offers []Offer
+ err = q.OffersByAddress(&offers, "GAXMF43TGZHW3QN3REOUA2U5PW5BTARXGGYJ3JIFHW3YT6QRKRL3CPPU", pq)
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(1, len(offers))
+ tt.Assert.True(offers[0].SellingAsset.Equals(xdr.MustNewNativeAsset()))
+ tt.Assert.True(offers[0].BuyingAsset.Equals(xdr.MustNewCreditAsset("USD", "GAXMF43TGZHW3QN3REOUA2U5PW5BTARXGGYJ3JIFHW3YT6QRKRL3CPPU")))
+ }
+
+ offers = []Offer{}
+ err = q.OffersByAddress(&offers, "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD", pq)
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(4, len(offers))
+
+ tt.Assert.Equal(4, len(offers))
+
+ tt.Assert.True(offers[0].SellingAsset.Equals(xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD")))
+ tt.Assert.True(offers[0].BuyingAsset.Equals(xdr.MustNewNativeAsset()))
+
+ tt.Assert.True(offers[1].SellingAsset.Equals(xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD")))
+ tt.Assert.True(offers[1].BuyingAsset.Equals(xdr.MustNewNativeAsset()))
+
+ tt.Assert.True(offers[2].SellingAsset.Equals(xdr.MustNewNativeAsset()))
+ tt.Assert.True(offers[2].BuyingAsset.Equals(xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD")))
+
+ tt.Assert.True(offers[3].SellingAsset.Equals(xdr.MustNewNativeAsset()))
+ tt.Assert.True(offers[3].BuyingAsset.Equals(xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD")))
+ }
+
+ var assets []xdr.Asset
+ err = q.ConnectedAssets(&assets, xdr.MustNewNativeAsset())
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(2, len(assets))
+ connectedAssetA := xdr.MustNewCreditAsset("USD", "GAXMF43TGZHW3QN3REOUA2U5PW5BTARXGGYJ3JIFHW3YT6QRKRL3CPPU")
+ connectedAssetB := xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD")
+ // It looks like there are some ordering changes between Postgres versions.
+ // We should really stick to a single version.
+ if assets[0].Equals(connectedAssetA) {
+ tt.Assert.True(assets[1].Equals(connectedAssetB), "%s %s", assets[0], assets[1])
+ }
+ if assets[0].Equals(connectedAssetB) {
+ tt.Assert.True(assets[1].Equals(connectedAssetA), "%s %s", assets[0], assets[1])
+ }
+ }
+
+ assets = []xdr.Asset{}
+ err = q.ConnectedAssets(&assets, xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD"))
+ if tt.Assert.NoError(err) {
+ tt.Assert.Equal(1, len(assets))
+ connectedAsset := xdr.MustNewNativeAsset()
+ tt.Assert.True(assets[0].Equals(connectedAsset))
+ }
+
+ var orderbookSummary OrderBookSummary
+ err = q.GetOrderBookSummary(&orderbookSummary, xdr.MustNewNativeAsset(), xdr.MustNewCreditAsset("USD", "GB2QIYT2IAUFMRXKLSLLPRECC6OCOGJMADSPTRK7TGNT2SFR2YGWDARD"), 10)
+ if tt.Assert.NoError(err) {
+ checkOrderBookRow(tt, orderbookSummary[0], "ask", int32(1), int32(1), float64(1), 300000000)
+ checkOrderBookRow(tt, orderbookSummary[1], "ask", int32(3), int32(1), float64(3), 400000000)
+
+ checkOrderBookRow(tt, orderbookSummary[2], "bid", int32(1), int32(2), float64(0.5), 200000000)
+ checkOrderBookRow(tt, orderbookSummary[3], "bid", int32(1), int32(1), float64(1), 100000000)
+ }
 }

services/horizon/internal/db2/core/offer.go

@@ -5,8 +5,8 @@ import (
  "math/big"
 
  sq "github.com/Masterminds/squirrel"
- "github.com/go-errors/errors"
  "github.com/stellar/go/services/horizon/internal/db2"
+ "github.com/stellar/go/support/errors"
  "github.com/stellar/go/xdr"
 )
 
@@ -24,7 +24,58 @@ func (r Offer) PriceAsString() string {
 // finding. Given the input asset type, a list of xdr.Assets is returned that
 // each have some available trades for the input asset.
 func (q *Q) ConnectedAssets(dest interface{}, selling xdr.Asset) error {
+ schemaVersion, err := q.SchemaVersion()
+ if err != nil {
+ return err
+ }
+
+ if schemaVersion < 9 {
+ return q.connectedAssetsSchema8(dest, selling)
+ } else {
+ return q.connectedAssetsSchema9(dest, selling)
+ }
+}
+
+func (q *Q) connectedAssetsSchema9(dest interface{}, selling xdr.Asset) error {
+ assets, ok := dest.(*[]xdr.Asset)
+ if !ok {
+ return errors.New("dest is not *[]xdr.Asset")
+ }
+
+ sellingAssetXDRString, err := xdr.MarshalBase64(selling)
+ if err != nil {
+ return errors.Wrap(err, "Error marshaling selling")
+ }
+
+ sql := sq.Select("buyingasset").
+ From("offers").
+ Where(sq.Eq{"sellingasset": sellingAssetXDRString}).
+ GroupBy("buyingasset")
+
+ var rows []struct {
+ Asset xdr.Asset `db:"buyingasset"`
+ }
+
+ err = q.Select(&rows, sql)
+
+ if err != nil {
+ return err
+ }
+
+ results := make([]xdr.Asset, len(rows))
+ *assets = results
+
+ for i, r := range rows {
+ results[i] = r.Asset
+ }
+
+ return nil
+}
 
+// ConnectedAssets loads xdr.Asset records for the purposes of path
+// finding. Given the input asset type, a list of xdr.Assets is returned that
+// each have some available trades for the input asset.
+func (q *Q) connectedAssetsSchema8(dest interface{}, selling xdr.Asset) error {
  assets, ok := dest.(*[]xdr.Asset)
  if !ok {
  return errors.New("dest is not *[]xdr.Asset")
@@ -81,6 +132,13 @@ func (q *Q) ConnectedAssets(dest interface{}, selling xdr.Asset) error {
 // OffersByAddress loads a page of active offers for the given
 // address.
 func (q *Q) OffersByAddress(dest interface{}, addy string, pq db2.PageQuery) error {
+ schemaVersion, err := q.SchemaVersion()
+ if err != nil {
+ return err
+ }
+
+ offers := []internalOffer{}
+
  sql := sq.Select("co.*").
  From("offers co").
  Where("co.sellerid = ?", addy).
@@ -98,5 +156,52 @@ func (q *Q) OffersByAddress(dest interface{}, addy string, pq db2.PageQuery) err
  sql = sql.Where("co.offerid < ?", cursor).OrderBy("co.offerid desc")
  }
 
- return q.Select(dest, sql)
+ err = q.Select(&offers, sql)
+ if err != nil {
+ return err
+ }
+
+ newOffers := make([]Offer, len(offers))
+
+ for i, offer := range offers {
+ newOffers[i] = offer.get()
+ }
+
+ if schemaVersion >= 9 {
+ *dest.(*[]Offer) = newOffers
+ return nil
+ }
+
+ // Convert schema 8 results to xdr.Assets
+ for i, offer := range offers {
+ var sellingAsset, buyingAsset xdr.Asset
+
+ if offer.SellingAssetType == xdr.AssetTypeAssetTypeNative {
+ sellingAsset.SetNative()
+ } else {
+ var account xdr.AccountId
+ err := account.SetAddress(offer.SellingIssuer.String)
+ if err != nil {
+ return errors.Wrap(err, "Error setting offer.SellingIssuer")
+ }
+ sellingAsset.SetCredit(offer.SellingAssetCode.String, account)
+ }
+
+ if offer.BuyingAssetType == xdr.AssetTypeAssetTypeNative {
+ buyingAsset.SetNative()
+ } else {
+ var account xdr.AccountId
+ err := account.SetAddress(offer.BuyingIssuer.String)
+ if err != nil {
+ return errors.Wrap(err, "Error setting offer.BuyingIssuer")
+ }
+ buyingAsset.SetCredit(offer.BuyingAssetCode.String, account)
+ }
+
+ newOffers[i].SellingAsset = sellingAsset
+ newOffers[i].BuyingAsset = buyingAsset
+ }
+
+ *dest.(*[]Offer) = newOffers
+ return nil
 }