feat: implement directional normalization for transactions

- Introduced a new `DirectionalNormalizer` interface to handle transaction normalization based on direction for both Stellar and XRP indexers.
- Added `normalizeDirectionalMetadata` function to process transaction metadata according to the specified direction.
- Updated `StellarIndexer` and `XRPIndexer` to implement the normalization logic.
- Enhanced tests to validate the correct handling of source-side routing for path payments in both Stellar and XRP transactions.
- Added new test cases to ensure proper functionality of the directional normalization feature.

Author: vietddude

internal/indexer/directional.go

@@ -0,0 +1,54 @@
+package indexer
+
+import (
+	"github.com/fystack/multichain-indexer/pkg/common/constant"
+	"github.com/fystack/multichain-indexer/pkg/common/types"
+)
+
+const (
+	metadataKeySourceAmount = "source_amount"
+	metadataKeySourceAsset  = "source_asset_address"
+	metadataKeySourceTxType = "source_tx_type"
+)
+
+func normalizeDirectionalMetadata(tx types.Transaction, direction string) types.Transaction {
+	if direction != types.DirectionOut {
+		clearDirectionalMetadata(&tx)
+		return tx
+	}
+
+	sourceType := tx.GetMetadataString(metadataKeySourceTxType)
+	if sourceType != "" {
+		tx.Type = constant.TxType(sourceType)
+		if tx.Type == constant.TxTypeNativeTransfer {
+			tx.AssetAddress = ""
+		}
+	}
+
+	if sourceAmount := tx.GetMetadataString(metadataKeySourceAmount); sourceAmount != "" {
+		tx.Amount = sourceAmount
+	}
+
+	switch tx.GetMetadataString(metadataKeySourceTxType) {
+	case string(constant.TxTypeNativeTransfer):
+		tx.AssetAddress = ""
+	case string(constant.TxTypeTokenTransfer):
+		tx.AssetAddress = tx.GetMetadataString(metadataKeySourceAsset)
+	}
+
+	clearDirectionalMetadata(&tx)
+	return tx
+}
+
+func clearDirectionalMetadata(tx *types.Transaction) {
+	if tx == nil || tx.Metadata == nil {
+		return
+	}
+
+	delete(tx.Metadata, metadataKeySourceAmount)
+	delete(tx.Metadata, metadataKeySourceAsset)
+	delete(tx.Metadata, metadataKeySourceTxType)
+	if len(tx.Metadata) == 0 {
+		tx.Metadata = nil
+	}
+}

internal/indexer/indexer.go

@@ -19,3 +19,11 @@ type Indexer interface {
 	GetBlocksByNumbers(ctx context.Context, blockNumbers []uint64) ([]BlockResult, error)
 	IsHealthy() bool
 }
+
+// DirectionalNormalizer is an optional hook for chains whose outgoing wallet
+// view differs from the incoming payload of the same routed transfer
+// (for example, path/cross-asset payments). It must not change from/to-based
+// routing decisions; it only shapes the emitted transaction payload.
+type DirectionalNormalizer interface {
+	NormalizeForDirection(tx types.Transaction, direction string) types.Transaction
+}

internal/indexer/stellar.go

@@ -53,6 +53,9 @@ func NewStellarIndexer(
 func (s *StellarIndexer) GetName() string                  { return strings.ToUpper(s.chainName) }
 func (s *StellarIndexer) GetNetworkType() enum.NetworkType { return enum.NetworkTypeStellar }
 func (s *StellarIndexer) GetNetworkInternalCode() string   { return s.config.InternalCode }
+func (s *StellarIndexer) NormalizeForDirection(tx types.Transaction, direction string) types.Transaction {
+	return normalizeDirectionalMetadata(tx, direction)
+}
 
 func (s *StellarIndexer) isMonitoredTransfer(from, to string) bool {
 	if s.pubkeyStore == nil {
@@ -490,6 +493,13 @@ func (s *StellarIndexer) convertPayment(
 	if memoType != "" {
 		tx.SetMetadata(types.MetadataKeyMemoType, memoType)
 	}
+	if sourceType, sourceAssetAddress, sourceAmount, ok := stellarSourcePaymentDetails(payment); ok {
+		tx.SetMetadata(metadataKeySourceTxType, string(sourceType))
+		tx.SetMetadata(metadataKeySourceAmount, sourceAmount)
+		if sourceType == constant.TxTypeTokenTransfer {
+			tx.SetMetadata(metadataKeySourceAsset, sourceAssetAddress)
+		}
+	}
 
 	return tx, true
 }
@@ -1021,6 +1031,28 @@ func isNativeStellarPayment(payment stellar.Payment) bool {
 	return strings.EqualFold(strings.TrimSpace(payment.AssetType), "native")
 }
 
+func stellarSourcePaymentDetails(payment stellar.Payment) (constant.TxType, string, string, bool) {
+	switch strings.ToLower(strings.TrimSpace(payment.Type)) {
+	case "path_payment_strict_receive", "path_payment_strict_recieve", "path_payment_strict_send":
+	default:
+		return "", "", "", false
+	}
+
+	amount := strings.TrimSpace(payment.SourceAmount)
+	if amount == "" {
+		return "", "", "", false
+	}
+	if strings.EqualFold(strings.TrimSpace(payment.SourceAssetType), "native") {
+		return constant.TxTypeNativeTransfer, "", amount, true
+	}
+
+	assetAddress := formatStellarAsset(payment.SourceAssetIssuer, payment.SourceAssetCode)
+	if assetAddress == "" {
+		return "", "", "", false
+	}
+	return constant.TxTypeTokenTransfer, assetAddress, amount, true
+}
+
 func stellarTransferFields(payment stellar.Payment) (string, string, string) {
 	switch strings.ToLower(strings.TrimSpace(payment.Type)) {
 	case "create_account":

internal/indexer/stellar_test.go

@@ -405,6 +405,54 @@ func TestStellarGetBlock_ParsesNativeAndIssuedPaymentsWithMemo(t *testing.T) {
 	assert.Equal(t, "0.00003", pathTx.TxFee.String())
 }
 
+func TestStellarConvertPayment_PathPaymentStoresSourceSideRouting(t *testing.T) {
+	t.Parallel()
+
+	idx := NewStellarIndexer(
+		"stellar_mainnet",
+		config.ChainConfig{NetworkId: "stellar_mainnet"},
+		nil,
+		nil,
+		mockStellarPubkeyStore{addresses: map[string]struct{}{"GDEST": {}}},
+	)
+
+	tx, ok := idx.convertPayment(
+		stellar.Payment{
+			Type:                  "path_payment_strict_send",
+			TransactionHash:       "tx-path",
+			TransactionSuccessful: true,
+			From:                  "GSOURCE",
+			To:                    "GDEST",
+			Amount:                "5.0000000",
+			AssetType:             "credit_alphanum4",
+			AssetCode:             "USDC",
+			AssetIssuer:           "GISSUER",
+			SourceAmount:          "10.0000000",
+			SourceAssetType:       "native",
+		},
+		nil,
+		21,
+		"ledger-hash",
+		"payment-1",
+		123,
+	)
+	require.True(t, ok)
+	assert.Equal(t, constant.TxTypeTokenTransfer, tx.Type)
+	assert.Equal(t, "GISSUER:USDC", tx.AssetAddress)
+	assert.Equal(t, "5.0000000", tx.Amount)
+	assert.Equal(t, string(constant.TxTypeNativeTransfer), tx.GetMetadataString(metadataKeySourceTxType))
+	assert.Equal(t, "10.0000000", tx.GetMetadataString(metadataKeySourceAmount))
+	assert.Equal(t, "", tx.GetMetadataString(metadataKeySourceAsset))
+
+	outTx := idx.NormalizeForDirection(tx, types.DirectionOut)
+	assert.Equal(t, constant.TxTypeNativeTransfer, outTx.Type)
+	assert.Equal(t, "", outTx.AssetAddress)
+	assert.Equal(t, "10.0000000", outTx.Amount)
+	assert.Equal(t, "", outTx.GetMetadataString(metadataKeySourceTxType))
+	assert.Equal(t, "", outTx.GetMetadataString(metadataKeySourceAmount))
+	assert.Equal(t, "", outTx.GetMetadataString(metadataKeySourceAsset))
+}
+
 func TestStellarGetBlock_FetchesTransactionDetailsLazily(t *testing.T) {
 	t.Parallel()
 
@@ -1019,6 +1067,19 @@ func TestStellarMainnetFetchAndParseTransactions(t *testing.T) {
 	}
 
 	testCases := []stellarRealTxCase{
+		{
+			name:          "native payment",
+			kind:          "payment",
+			txHash:        "81dba977244285fd3a6e9192ff2a594fe99d9bd6ec892533fe80c5f25c77c1f0",
+			transferIndex: "265365546521460737",
+			wantType:      constant.TxTypeNativeTransfer,
+			wantFrom:      "GBC6NRTTQLRCABQHIR5J4R4YDJWFWRAO4ZRQIM2SVI5GSIZ2HZ42RINW",
+			wantTo:        "GCZNOTQRRETQLBQH2MPWYMCLQBYMXKZI7XXYHS7F5RJHH7VMATQ57TQZ",
+			wantAmount:    "120.0621870",
+			verify: func(t *testing.T, tx types.Transaction, _ *StellarIndexer, _ *stellar.Payment, _ *stellar.Operation, _ []stellar.Effect) {
+				assert.Equal(t, "", tx.AssetAddress)
+			},
+		},
 		{
 			name:          "create account",
 			kind:          "payment",