#4728: include contract asset balance changes in payments

protocols/horizon/operations/main.go

@@ -352,9 +352,10 @@ type LiquidityPoolWithdraw struct {
 // just contains a base64 encoded string of it's xdr serialization.
 type InvokeHostFunction struct {
 	Base
-	Parameters []HostFunctionParameter `json:"parameters"`
-	Function   string                  `json:"function"`
-	Footprint  string                  `json:"footprint"`
+	Parameters          []HostFunctionParameter      `json:"parameters"`
+	Function            string                       `json:"function"`
+	Footprint           string                       `json:"footprint"`
+	AssetBalanceChanges []AssetContractBalanceChange `json:"asset_balance_changes"`
 }
 
 // InvokeHostFunction parameter model, intentionally simplified, Value
@@ -364,6 +365,17 @@ type HostFunctionParameter struct {
 	Type  string `json:"type"`
 }
 
+type AssetContractBalanceChange struct {
+	base.Asset
+	From        string `json:"from"`
+	FromMuxed   string `json:"from_muxed,omitempty"`
+	FromMuxedID uint64 `json:"from_muxed_id,omitempty,string"`
+	To          string `json:"to"`
+	ToMuxed     string `json:"to_muxed,omitempty"`
+	ToMuxedID   uint64 `json:"to_muxed_id,omitempty,string"`
+	Amount      string `json:"amount"`
+}
+
 // Operation interface contains methods implemented by the operation types
 type Operation interface {
 	GetBase() Base

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

@@ -648,6 +648,7 @@ type Operation struct {
 	SourceAccount         string            `db:"source_account"`
 	SourceAccountMuxed    null.String       `db:"source_account_muxed"`
 	TransactionSuccessful bool              `db:"transaction_successful"`
+	AssetBalanceChanged   bool              `db:"asset_balance_changed"`
 }
 
 // ManageOffer is a struct of data from `operations.DetailsString`

services/horizon/internal/db2/history/operation.go

@@ -236,15 +236,18 @@ func (q *OperationsQ) ForTransaction(ctx context.Context, hash string) *Operatio
 
 // OnlyPayments filters the query being built to only include operations that
 // are in the "payment" class of operations:  CreateAccountOps, Payments, and
-// PathPayments.
+// PathPayments and contract asset balance changes expressed in 'asset_balance_changed' flag.
 func (q *OperationsQ) OnlyPayments() *OperationsQ {
-	q.sql = q.sql.Where(sq.Eq{"hop.type": []xdr.OperationType{
-		xdr.OperationTypeCreateAccount,
-		xdr.OperationTypePayment,
-		xdr.OperationTypePathPaymentStrictReceive,
-		xdr.OperationTypePathPaymentStrictSend,
-		xdr.OperationTypeAccountMerge,
-	}})
+	q.sql = q.sql.Where(sq.Or{
+		sq.Eq{"hop.type": []xdr.OperationType{
+			xdr.OperationTypeCreateAccount,
+			xdr.OperationTypePayment,
+			xdr.OperationTypePathPaymentStrictReceive,
+			xdr.OperationTypePathPaymentStrictSend,
+			xdr.OperationTypeAccountMerge,
+		}},
+		sq.Eq{"hop.asset_balance_changed": 1}})
+
 	return q
 }
 
@@ -390,6 +393,7 @@ var selectOperation = sq.Select(
 		"hop.details, " +
 		"hop.source_account, " +
 		"hop.source_account_muxed, " +
+		"hop.asset_balance_changed, " +
 		"ht.transaction_hash, " +
 		"ht.tx_result, " +
 		"COALESCE(ht.successful, true) as transaction_successful").

services/horizon/internal/db2/history/operation_batch_insert_builder.go

@@ -20,6 +20,7 @@ type OperationBatchInsertBuilder interface {
 		details []byte,
 		sourceAccount string,
 		sourceAcccountMuxed null.String,
+		assetBalanceChanged bool,
 	) error
 	Exec(ctx context.Context) error
 }
@@ -49,15 +50,17 @@ func (i *operationBatchInsertBuilder) Add(
 	details []byte,
 	sourceAccount string,
 	sourceAccountMuxed null.String,
+	assetBalanceChanged bool,
 ) error {
 	return i.builder.Row(ctx, map[string]interface{}{
-		"id":                   id,
-		"transaction_id":       transactionID,
-		"application_order":    applicationOrder,
-		"type":                 operationType,
-		"details":              details,
-		"source_account":       sourceAccount,
-		"source_account_muxed": sourceAccountMuxed,
+		"id":                    id,
+		"transaction_id":        transactionID,
+		"application_order":     applicationOrder,
+		"type":                  operationType,
+		"details":               details,
+		"source_account":        sourceAccount,
+		"source_account_muxed":  sourceAccountMuxed,
+		"asset_balance_changed": assetBalanceChanged,
 	})
 
 }

services/horizon/internal/db2/history/operation_batch_insert_builder_test.go

@@ -56,6 +56,7 @@ func TestAddOperation(t *testing.T) {
 		details,
 		sourceAccount,
 		null.StringFrom(sourceAccountMuxed),
+		true,
 	)
 	tt.Assert.NoError(err)
 

services/horizon/internal/db2/history/operation_test.go

@@ -103,6 +103,7 @@ func TestOperationByLiquidityPool(t *testing.T) {
 		[]byte("{}"),
 		"GAUJETIZVEP2NRYLUESJ3LS66NVCEGMON4UDCBCSBEVPIID773P2W6AY",
 		null.String{},
+		false,
 	)
 	tt.Assert.NoError(err)
 	err = operationBuilder.Exec(tt.Ctx)
@@ -117,6 +118,7 @@ func TestOperationByLiquidityPool(t *testing.T) {
 		[]byte("{}"),
 		"GAUJETIZVEP2NRYLUESJ3LS66NVCEGMON4UDCBCSBEVPIID773P2W6AY",
 		null.String{},
+		false,
 	)
 	tt.Assert.NoError(err)
 	err = operationBuilder.Exec(tt.Ctx)

services/horizon/internal/db2/schema/migrations/64_add_payment_flag_history_ops.sql

@@ -0,0 +1,7 @@
+-- +migrate Up
+
+ALTER TABLE history_operations ADD asset_balance_changed smallint DEFAULT 0;
+
+-- +migrate Down
+
+ALTER TABLE history_operations DROP COLUMN asset_balance_changed;

services/horizon/internal/ingest/processor_runner.go

@@ -145,7 +145,7 @@ func (s *ProcessorRunner) buildTransactionProcessor(
 		statsLedgerTransactionProcessor,
 		processors.NewEffectProcessor(s.historyQ, sequence),
 		processors.NewLedgerProcessor(s.historyQ, ledger, CurrentVersion),
-		processors.NewOperationProcessor(s.historyQ, sequence),
+		processors.NewOperationProcessor(s.historyQ, sequence, s.config.NetworkPassphrase),
 		tradeProcessor,
 		processors.NewParticipantsProcessor(s.historyQ, sequence),
 		processors.NewTransactionProcessor(s.historyQ, sequence),

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

@@ -11,6 +11,7 @@ import (
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/protocols/horizon/base"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
+	"github.com/stellar/go/support/contractevents"
 	"github.com/stellar/go/support/errors"
 	"github.com/stellar/go/toid"
 	"github.com/stellar/go/xdr"
@@ -20,26 +21,29 @@ import (
 type OperationProcessor struct {
 	operationsQ history.QOperations
 
-	sequence uint32
-	batch    history.OperationBatchInsertBuilder
+	sequence          uint32
+	batch             history.OperationBatchInsertBuilder
+	networkPassphrase string
 }
 
-func NewOperationProcessor(operationsQ history.QOperations, sequence uint32) *OperationProcessor {
+func NewOperationProcessor(operationsQ history.QOperations, sequence uint32, networkPassphrase string) *OperationProcessor {
 	return &OperationProcessor{
-		operationsQ: operationsQ,
-		sequence:    sequence,
-		batch:       operationsQ.NewOperationBatchInsertBuilder(maxBatchSize),
+		operationsQ:       operationsQ,
+		sequence:          sequence,
+		batch:             operationsQ.NewOperationBatchInsertBuilder(maxBatchSize),
+		networkPassphrase: networkPassphrase,
 	}
 }
 
 // ProcessTransaction process the given transaction
 func (p *OperationProcessor) ProcessTransaction(ctx context.Context, transaction ingest.LedgerTransaction) error {
 	for i, op := range transaction.Envelope.Operations() {
 		operation := transactionOperationWrapper{
-			index:          uint32(i),
-			transaction:    transaction,
-			operation:      op,
-			ledgerSequence: p.sequence,
+			index:             uint32(i),
+			transaction:       transaction,
+			operation:         op,
+			ledgerSequence:    p.sequence,
+			networkPassphrase: p.networkPassphrase,
 		}
 		details, err := operation.Details()
 		if err != nil {
@@ -65,6 +69,7 @@ func (p *OperationProcessor) ProcessTransaction(ctx context.Context, transaction
 			detailsJSON,
 			acID.Address(),
 			sourceAccountMuxed,
+			operation.AssetBalanceChanged(),
 		); err != nil {
 			return errors.Wrap(err, "Error batch inserting operation rows")
 		}
@@ -79,10 +84,11 @@ func (p *OperationProcessor) Commit(ctx context.Context) error {
 
 // transactionOperationWrapper represents the data for a single operation within a transaction
 type transactionOperationWrapper struct {
-	index          uint32
-	transaction    ingest.LedgerTransaction
-	operation      xdr.Operation
-	ledgerSequence uint32
+	index             uint32
+	transaction       ingest.LedgerTransaction
+	operation         xdr.Operation
+	ledgerSequence    uint32
+	networkPassphrase string
 }
 
 // ID returns the ID for the operation.
@@ -248,6 +254,44 @@ func (operation *transactionOperationWrapper) OperationResult() *xdr.OperationRe
 	return &tr
 }
 
+func (operation *transactionOperationWrapper) AssetBalanceChanged() bool {
+	switch operation.OperationType() {
+	case xdr.OperationTypeCreateAccount:
+		return true
+	case xdr.OperationTypePayment:
+		return true
+	case xdr.OperationTypePathPaymentStrictReceive:
+		return true
+	case xdr.OperationTypePathPaymentStrictSend:
+		return true
+	case xdr.OperationTypeAccountMerge:
+		return true
+	case xdr.OperationTypeInvokeHostFunction:
+		txMeta, ok := operation.transaction.UnsafeMeta.GetV3()
+		if !ok {
+			return false
+		}
+		for _, opEvents := range txMeta.Events {
+			for _, contractEvent := range opEvents.Events {
+				if sacEvent, err := contractevents.NewStellarAssetContractEvent(&contractEvent, operation.networkPassphrase); err == nil {
+					switch sacEvent.GetType() {
+					case contractevents.EventTypeTransfer:
+						return true
+					case contractevents.EventTypeMint:
+						return true
+					case contractevents.EventTypeClawback:
+						return true
+					case contractevents.EventTypeBurn:
+						return true
+					}
+				}
+			}
+		}
+	}
+
+	return false
+}
+
 func (operation *transactionOperationWrapper) findInitatingBeginSponsoringOp() *transactionOperationWrapper {
 	if !operation.transaction.Result.Successful() {
 		// Failed transactions may not have a compliant sandwich structure
@@ -605,8 +649,14 @@ func (operation *transactionOperationWrapper) Details() (map[string]interface{},
 				}
 				params = append(params, serializedParam)
 			}
-
 			details["parameters"] = params
+
+			if balanceChanges, err := operation.parseAssetBalanceChangesFromContractEvents(); err != nil {
+				return nil, err
+			} else {
+				details["asset_balance_changes"] = balanceChanges
+			}
+
 		case xdr.HostFunctionTypeHostFunctionTypeCreateContract:
 			args := op.Function.MustCreateContractArgs()
 			details["type"] = args.ContractId.Type.String()
@@ -655,6 +705,64 @@ func (operation *transactionOperationWrapper) Details() (map[string]interface{},
 	return details, nil
 }
 
+// Searches an operation for SAC events that are of a type which represent asset balances changed between contracts and/or classic.
+// SAC events have a one-to-one association to SAC contract fn invocations, i.e. invoke the 'mint' function, will trigger one Mint Event to be emitted capturing the fn args.
+// SAC events that involve asset balance changes follow some standard data formats:
+//     amount expressed as uint64 only, the event type in this case provides the context of whether an amount was credit/debit to a balance.
+//     the 'from' and 'to' attributes represent an account or a contract
+func (operation *transactionOperationWrapper) parseAssetBalanceChangesFromContractEvents() ([]map[string]interface{}, error) {
+	balanceChanges := []map[string]interface{}{}
+
+	txMeta, ok := operation.transaction.UnsafeMeta.GetV3()
+	if !ok {
+		// there's no newer v3 meta which is when contract events start being present, otherwise none present.
+		return balanceChanges, nil
+	}
+
+	for _, opEvents := range txMeta.Events {
+		for _, contractEvent := range opEvents.Events {
+			// parse the xdr contract event to contractevents.StellarAssetContractEvent model
+			// has some convenience like to/from attributes are expressed in strkey format for contracts(C...) and accounts(G...)
+			if sacEvent, err := contractevents.NewStellarAssetContractEvent(&contractEvent, operation.networkPassphrase); err == nil {
+				switch sacEvent.GetType() {
+				case contractevents.EventTypeTransfer:
+					transferEvt := sacEvent.(contractevents.TransferEvent)
+					balanceChanges = append(balanceChanges, createSACBalanceChangeEntry(transferEvt.From, transferEvt.To, transferEvt.Amount.Lo, transferEvt.Asset))
+				case contractevents.EventTypeMint:
+					mintEvt := sacEvent.(contractevents.MintEvent)
+					balanceChanges = append(balanceChanges, createSACBalanceChangeEntry(mintEvt.Admin, mintEvt.To, mintEvt.Amount.Lo, mintEvt.Asset))
+				case contractevents.EventTypeClawback:
+					clawbackEvt := sacEvent.(contractevents.ClawbackEvent)
+					balanceChanges = append(balanceChanges, createSACBalanceChangeEntry(clawbackEvt.From, clawbackEvt.Admin, clawbackEvt.Amount.Lo, clawbackEvt.Asset))
+				case contractevents.EventTypeBurn:
+					burnEvt := sacEvent.(contractevents.BurnEvent)
+					balanceChanges = append(balanceChanges, createSACBalanceChangeEntry(burnEvt.From, "", burnEvt.Amount.Lo, burnEvt.Asset))
+				}
+			}
+		}
+	}
+
+	return balanceChanges, nil
+}
+
+// fromAccount - strkey format of contract or address
+// toAccount - strkey format of contract or address, or nillable
+// amountChanged - absolute value that asset balance changed
+// asset - the fully qualified issuer:code for asset that had balance change
+//
+// return - a balance changed record expressed as map of key/value's
+func createSACBalanceChangeEntry(fromAccount string, toAccount string, amountChanged xdr.Uint64, asset xdr.Asset) map[string]interface{} {
+	balanceChange := map[string]interface{}{}
+
+	balanceChange["from"] = fromAccount
+	if toAccount != "" {
+		balanceChange["to"] = toAccount
+	}
+	balanceChange["amount"] = string(amountChanged)
+	addAssetDetails(balanceChange, asset, "")
+	return balanceChange
+}
+
 func addLiquidityPoolAssetDetails(result map[string]interface{}, lpp xdr.LiquidityPoolParameters) error {
 	result["asset_type"] = "liquidity_pool_shares"
 	if lpp.Type != xdr.LiquidityPoolTypeLiquidityPoolConstantProduct {

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

@@ -41,6 +41,7 @@ func (s *OperationsProcessorTestSuiteLedger) SetupTest() {
 	s.processor = NewOperationProcessor(
 		s.mockQ,
 		56,
+		"test network",
 	)
 }
 

support/contractevents/event.go

@@ -48,11 +48,11 @@ type sacEvent struct {
 	Asset xdr.Asset
 }
 
-func (e *sacEvent) GetAsset() xdr.Asset {
+func (e sacEvent) GetAsset() xdr.Asset {
 	return e.Asset
 }
 
-func (e *sacEvent) GetType() EventType {
+func (e sacEvent) GetType() EventType {
 	return e.Type
 }