horizon/ingest: Add method to ingest.Change that returns liquidity po…

…ol type (#4158)

* Add method to ingest.Change that returns liquidity pool type

ingest/change.go

@@ -82,6 +82,30 @@ func (c *Change) LedgerEntryChangeType() xdr.LedgerEntryChangeType {
  }
 }
 
+// getLiquidityPool gets the most recent state of the LiquidityPool that exists or existed.
+func (c *Change) getLiquidityPool() (*xdr.LiquidityPoolEntry, error) {
+ var entry *xdr.LiquidityPoolEntry
+ if c.Pre != nil {
+ entry = c.Pre.Data.LiquidityPool
+ }
+ if c.Post != nil {
+ entry = c.Post.Data.LiquidityPool
+ }
+ if entry == nil {
+ return &xdr.LiquidityPoolEntry{}, errors.New("this change does not include a liquidity pool")
+ }
+ return entry, nil
+}
+
+// GetLiquidityPoolType returns the liquidity pool type.
+func (c *Change) GetLiquidityPoolType() (xdr.LiquidityPoolType, error) {
+ lp, err := c.getLiquidityPool()
+ if err != nil {
+ return xdr.LiquidityPoolType(0), err
+ }
+ return lp.Body.Type, nil
+}
+
 // AccountChangedExceptSigners returns true if account has changed WITHOUT
 // checking the signers (except master key weight!). In other words, if the only
 // change is connected to signers, this function will return false.

services/horizon/internal/ingest/processors/asset_stats_set.go

@@ -237,12 +237,9 @@ func (s AssetStatSet) AddLiquidityPool(change ingest.Change) error {
  return ingest.NewStateError(errors.New("both pre and post liquidity pools cannot be nil"))
  }
 
- var lpType xdr.LiquidityPoolType
- if pre != nil {
- lpType = pre.Body.Type
- }
- if post != nil {
- lpType = post.Body.Type
+ lpType, err := change.GetLiquidityPoolType()
+ if err != nil {
+ return ingest.NewStateError(err)
  }
 
  var assetA, assetB xdr.Asset