Send account_merge operations to callbacks.receive

src/github.com/stellar/gateway/horizon/effect_response.go

@@ -0,0 +1,14 @@
+package horizon
+
+// EffectsPageResponse contains page of effects returned by Horizon
+type EffectsPageResponse struct {
+	Embedded struct {
+		Records []EffectResponse
+	} `json:"_embedded"`
+}
+
+// EffectResponse contains effect data returned by Horizon
+type EffectResponse struct {
+	Type   string `json:"type"`
+	Amount string `json:"amount"`
+}

src/github.com/stellar/gateway/horizon/main.go

@@ -13,6 +13,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/stellar/go/support/errors"
 	"github.com/stellar/go/xdr"
 )
 
@@ -23,6 +24,7 @@ type PaymentHandler func(PaymentResponse) error
 type HorizonInterface interface {
 	LoadAccount(accountID string) (response AccountResponse, err error)
 	LoadMemo(p *PaymentResponse) (err error)
+	LoadAccountMergeAmount(p *PaymentResponse) error
 	LoadOperation(operationID string) (response PaymentResponse, err error)
 	StreamPayments(accountID string, cursor *string, onPaymentHandler PaymentHandler) (err error)
 	SubmitTransaction(txeBase64 string) (response SubmitTransactionResponse, err error)
@@ -125,6 +127,33 @@ func (h *Horizon) LoadMemo(p *PaymentResponse) (err error) {
 	return json.NewDecoder(res.Body).Decode(&p.Memo)
 }
 
+// LoadAccountMergeAmount loads `account_merge` operation amount from it's effects
+func (h *Horizon) LoadAccountMergeAmount(p *PaymentResponse) error {
+	if p.Type != "account_merge" {
+		return errors.New("Not `account_merge` operation")
+	}
+
+	res, err := http.Get(p.Links.Effects.Href)
+	if err != nil {
+		return errors.Wrap(err, "Error getting effects for operation")
+	}
+	defer res.Body.Close()
+	var page EffectsPageResponse
+	err = json.NewDecoder(res.Body).Decode(&page)
+	if err != nil {
+		return errors.Wrap(err, "Error decoding effects page")
+	}
+
+	for _, effect := range page.Embedded.Records {
+		if effect.Type == "account_credited" {
+			p.Amount = effect.Amount
+			return nil
+		}
+	}
+
+	return errors.New("Could not find `account_credited` effect in `account_merge` operation effects")
+}
+
 // StreamPayments streams incoming payments
 func (h *Horizon) StreamPayments(accountID string, cursor *string, onPaymentHandler PaymentHandler) (err error) {
 	url := h.ServerURL + "/accounts/" + accountID + "/payments"

src/github.com/stellar/gateway/horizon/payment_response.go

@@ -10,6 +10,9 @@ type PaymentResponse struct {
 		Transaction struct {
 			Href string `json:"href"`
 		} `json:"transaction"`
+		Effects struct {
+			Href string `json:"href"`
+		} `json:"effects"`
 	} `json:"_links"`
 
 	// payment/path_payment fields
@@ -20,6 +23,10 @@ type PaymentResponse struct {
 	AssetIssuer string `json:"asset_issuer"`
 	Amount      string `json:"amount"`
 
+	// account_merge
+	Account string `json:"account"`
+	Into    string `json:"into"`
+
 	// transaction fields
 	Memo struct {
 		Type  string `json:"memo_type"`

src/github.com/stellar/gateway/listener/payment_listener.go

@@ -213,11 +213,15 @@ func (pl *PaymentListener) onPayment(payment horizon.PaymentResponse) (err error
 // shouldProcessPayment returns false and text status if payment should not be processed
 // (ex. asset is different than allowed assets).
 func (pl *PaymentListener) shouldProcessPayment(payment horizon.PaymentResponse) (bool, string) {
-	if payment.Type != "payment" && payment.Type != "path_payment" {
+	if payment.Type != "payment" && payment.Type != "path_payment" && payment.Type != "account_merge" {
 		return false, "Not a payment operation"
 	}
 
-	if payment.To != pl.config.Accounts.ReceivingAccountID {
+	if payment.Type == "account_merge" {
+		payment.AssetType = "native"
+	}
+
+	if payment.To != pl.config.Accounts.ReceivingAccountID && payment.Into != pl.config.Accounts.ReceivingAccountID {
 		return false, "Operation sent not received"
 	}
 
@@ -229,6 +233,17 @@ func (pl *PaymentListener) shouldProcessPayment(payment horizon.PaymentResponse)
 }
 
 func (pl *PaymentListener) process(payment horizon.PaymentResponse) error {
+	if payment.Type == "account_merge" {
+		payment.AssetType = "native"
+		payment.From = payment.Account
+		payment.To = payment.Into
+
+		err := pl.horizon.LoadAccountMergeAmount(&payment)
+		if err != nil {
+			return errors.Wrap(err, "Unable to load account_merge amount")
+		}
+	}
+
 	err := pl.horizon.LoadMemo(&payment)
 	if err != nil {
 		return errors.Wrap(err, "Unable to load transaction memo")

src/github.com/stellar/gateway/listener/payment_listener_test.go

@@ -347,6 +347,60 @@ func TestPaymentListener(t *testing.T) {
 			})
 		})
 
+		Convey("When receive callback returns success (account_merge)", func() {
+			operation.Type = "account_merge"
+			operation.Account = "GBL27BKG2JSDU6KQ5YJKCDWTVIU24VTG4PLB63SF4K2DBZS5XZMWRPVU"
+			operation.Into = "GATKP6ZQM5CSLECPMTAC5226PE367QALCPM6AFHTSULPPZMT62OOPMQB"
+			operation.Amount = "100"
+			operation.Memo.Type = "text"
+			operation.Memo.Value = "testing"
+
+			// Updated in the listener
+			operation.From = "GBL27BKG2JSDU6KQ5YJKCDWTVIU24VTG4PLB63SF4K2DBZS5XZMWRPVU"
+			operation.To = "GATKP6ZQM5CSLECPMTAC5226PE367QALCPM6AFHTSULPPZMT62OOPMQB"
+			operation.AssetType = "native"
+
+			config.Assets[1].Code = "XLM"
+			config.Assets[1].Issuer = ""
+
+			mockRepository.On("GetReceivedPaymentByOperationID", int64(1)).Return(nil, nil).Once()
+			mockHorizon.On("LoadAccountMergeAmount", &operation).Return(nil).Once()
+			mockHorizon.On("LoadMemo", &operation).Return(nil).Once()
+
+			mockEntityManager.On("Persist", mock.AnythingOfType("*entities.ReceivedPayment")).
+				Run(ensurePaymentStatus(t, operation, "Processing...")).Return(nil).Once()
+
+			mockEntityManager.On("Persist", mock.AnythingOfType("*entities.ReceivedPayment")).
+				Run(ensurePaymentStatus(t, operation, "Success")).Return(nil).Once()
+
+			mockHTTPClient.On(
+				"Do",
+				mock.MatchedBy(func(req *http.Request) bool {
+					return req.URL.String() == "http://receive_callback"
+				}),
+			).Return(
+				net.BuildHTTPResponse(200, "ok"),
+				nil,
+			).Run(func(args mock.Arguments) {
+				req := args.Get(0).(*http.Request)
+
+				assert.Equal(t, operation.Account, req.PostFormValue("from"))
+				assert.Equal(t, operation.Amount, req.PostFormValue("amount"))
+				assert.Equal(t, operation.AssetCode, req.PostFormValue("asset_code"))
+				assert.Equal(t, operation.AssetIssuer, req.PostFormValue("asset_issuer"))
+				assert.Equal(t, operation.Memo.Type, req.PostFormValue("memo_type"))
+				assert.Equal(t, operation.Memo.Value, req.PostFormValue("memo"))
+			}).Once()
+
+			Convey("it should save the status", func() {
+				err := paymentListener.onPayment(operation)
+				assert.Nil(t, err)
+				mockHorizon.AssertExpectations(t)
+				mockEntityManager.AssertExpectations(t)
+				mockRepository.AssertExpectations(t)
+			})
+		})
+
 		Convey("When receive callback returns success (no memo)", func() {
 			operation.Type = "payment"
 			operation.To = "GATKP6ZQM5CSLECPMTAC5226PE367QALCPM6AFHTSULPPZMT62OOPMQB"

src/github.com/stellar/gateway/mocks/main.go

@@ -102,6 +102,12 @@ func (m *MockHorizon) LoadMemo(p *horizon.PaymentResponse) (err error) {
 	return a.Error(0)
 }
 
+// LoadAccountMergeAmount is a mocking a method
+func (m *MockHorizon) LoadAccountMergeAmount(p *horizon.PaymentResponse) (err error) {
+	a := m.Called(p)
+	return a.Error(0)
+}
+
 // StreamPayments is a mocking a method
 func (m *MockHorizon) StreamPayments(accountID string, cursor *string, onPaymentHandler horizon.PaymentHandler) (err error) {
 	a := m.Called(accountID, cursor, onPaymentHandler)