dflow parser

cmd/shreder/main.go

@@ -55,6 +55,11 @@ func main() {
 					"proVF4pMXVaYqmy4NjniPh4pqKNfMmsihgd4wdkCX3u",
 				},
 			},
+			"dflow": {
+				AccountRequired: []string{
+					"DF1ow4tspfHX9JwWJsAb9epbkA8hmpSEAtxXy1V27QBH",
+				},
+			},
 			// TODO: axiom, gmgn, etc.
 		})
 	if err != nil {
@@ -89,12 +94,8 @@ func main() {
 		case txBatch := <-txCh:
 			//jsonData, _ := json.MarshalIndent(txBatch, "", "  ")
 			for _, tx := range txBatch {
-				if tx.Label == "okxdexroutev2" {
+				if tx.Label == "dflow" {
-					if tx.Event == "buy" {
+					fmt.Println("===============", tx.TxHash, tx.Event, tx.Token0Address, "token:", tx.Token0Amount)
-						fmt.Println("===============", tx.TxHash, tx.Event, tx.Token0Address, "token:", tx.Token0Amount, "sol:", tx.Token1Amount)
-					} else if tx.Event == "sell" {
-						fmt.Println("===============", tx.TxHash, tx.Event, tx.Token0Address, "token:", tx.Token0Amount)
-					}
 				}
 			}
 			//fmt.Println(txBatch[0].TxHash)

pkg/shreder/dflow.go

@@ -0,0 +1,323 @@
+package shreder
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+
+	bin "github.com/gagliardetto/binary"
+	"github.com/gagliardetto/solana-go"
+	"github.com/shopspring/decimal"
+)
+
+var (
+	dflowProgramID = solana.MustPublicKeyFromBase58("DF1ow4tspfHX9JwWJsAb9epbkA8hmpSEAtxXy1V27QBH")
+
+	dflowSwapDisc                   = []byte{248, 198, 158, 145, 225, 117, 135, 200}
+	dflowSwap2Disc                  = []byte{65, 75, 63, 76, 235, 91, 91, 136}
+	dflowSwapWithDestinationDisc    = []byte{168, 172, 24, 77, 197, 156, 135, 101}
+	dflowSwapWithDestinationNative  = []byte{205, 77, 127, 108, 241, 32, 196, 195}
+	dflowSwap2WithDestinationDisc   = []byte{95, 123, 213, 246, 122, 1, 86, 231}
+	dflowSwap2WithDestinationNative = []byte{222, 100, 184, 146, 186, 196, 105, 165}
+
+	wrappedSOL = solana.MustPublicKeyFromBase58("So11111111111111111111111111111111111111112")
+)
+
+// Action enum tags (0-based, per dflow_idl Action variants)
+const (
+	ActWhirlpoolsSwap uint8 = iota
+	ActClearpoolsSwap
+	ActRaydiumAmmSwap
+	ActLifinityV2Swap
+	ActMeteoraDlmmSwap
+	ActRaydiumClmmSwap
+	ActRaydiumClmmSwapV2
+	ActPhoenixSwap
+	ActPumpFunBuy
+	ActPumpFunSell
+	ActGammaSwap
+	ActObricV2Swap
+	ActPumpFunAmmBuy
+	ActPumpFunAmmSell
+	ActSolFiSwap
+	ActRubiconSwap
+	ActMeteoraDammV1Swap
+	ActRaydiumCpSwap
+	ActStabbleStableSwap
+	ActTesseraVSwap
+	ActMeteoraDammV2Swap
+	ActRaydiumLaunchlabSwap
+	ActMeteoraDbcSwap
+	ActHumidiFiSwap
+	ActWhirlpoolsSwapV2
+	ActMeteoraDlmmSwapV2
+	ActZeroFiSwap
+	ActAlphaQSwap
+	ActTokenSwap
+	ActSolFiV2Swap
+	ActMozartSwap
+	ActDFlowDynamicRouteV1
+	ActHeavenSwap
+	ActNexusSwap
+	ActSarosDlmmSwap
+	ActTransferFee
+	ActTransferFeeWithMint
+	ActRecordId
+	ActRecordId2
+	ActManifestSwap
+	ActBisonFiSwap
+	ActSanctumInfinitySwap
+	ActSanctumInfinityLiquidity
+	ActOpenPredictionsOrder
+	ActScorchSwap
+	ActIncludeAccount
+)
+
+// DynamicRouteV1CandidateAction tags
+const (
+	drv1SolFi uint8 = iota
+	drv1Rubicon
+	drv1TesseraV
+	drv1HumidiFi
+	drv1SolFiV2
+	drv1Mozart
+	drv1ObricV2
+	drv1Nexus
+)
+
+// PumpFunAmmSellOptions { amount: u64, orchestrator_flags: OrchestratorFlags{flags u8} }
+type pumpFunAmm struct {
+	Amount uint64
+	Flags  uint8
+}
+
+type dflowAction struct {
+	Tag  uint8
+	Pump *pumpFunAmm
+}
+
+type dflowSwapParams struct {
+	Actions []dflowAction
+}
+
+// bytes to skip for Action variants before/after PumpFunAmmSell; only PumpFunAmmSell is decoded.
+func skipDflowAction(dec *bin.Decoder, tag uint8) (*pumpFunAmm, error) {
+	switch tag {
+	case ActWhirlpoolsSwap, ActClearpoolsSwap, ActWhirlpoolsSwapV2:
+		// amount u64 + bool + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActRaydiumAmmSwap, ActLifinityV2Swap, ActPumpFunBuy, ActPumpFunSell, ActObricV2Swap,
+		ActSolFiSwap, ActRubiconSwap, ActMeteoraDammV1Swap, ActRaydiumCpSwap,
+		ActStabbleStableSwap, ActTesseraVSwap, ActMeteoraDammV2Swap, ActRaydiumLaunchlabSwap,
+		ActZeroFiSwap, ActAlphaQSwap, ActTokenSwap, ActSolFiV2Swap, ActMozartSwap, ActHeavenSwap,
+		ActNexusSwap, ActSarosDlmmSwap, ActManifestSwap, ActBisonFiSwap:
+		// amount u64 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1)
+	case ActMeteoraDlmmSwap, ActRaydiumClmmSwap, ActRaydiumClmmSwapV2, ActMeteoraDlmmSwapV2:
+		// amount u64 + u8 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActPhoenixSwap:
+		// amount u64 + side u8 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActGammaSwap:
+		// amount u64 + endorsed bool + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActPumpFunAmmSell, ActPumpFunAmmBuy:
+		amt, err := dec.ReadUint64(binary.LittleEndian)
+		if err != nil {
+			return nil, err
+		}
+		flg, err := dec.ReadUint8()
+		if err != nil {
+			return nil, err
+		}
+		return &pumpFunAmm{Amount: amt, Flags: flg}, nil
+	case ActMeteoraDbcSwap:
+		// amount u64 + is_rate_limiter_applied bool + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActHumidiFiSwap:
+		// amount u64 + swap_id u64 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 8 + 1)
+	case ActDFlowDynamicRouteV1:
+		// candidate_actions Vec<DynamicRouteV1CandidateAction> + amount u64 + orchestrator_flags u8
+		ln, err := dec.ReadUint32(binary.LittleEndian)
+		if err != nil {
+			return nil, err
+		}
+		for j := uint32(0); j < ln; j++ {
+			t, err := dec.ReadUint8()
+			if err != nil {
+				return nil, err
+			}
+			if t == drv1HumidiFi {
+				if err := dec.SkipBytes(8); err != nil {
+					return nil, err
+				}
+			}
+			// other variants carry no payload
+		}
+		if err := dec.SkipBytes(8); err != nil { // amount
+			return nil, err
+		}
+		return nil, dec.SkipBytes(1) // orchestrator_flags
+	case ActTransferFee, ActTransferFeeWithMint:
+		return nil, dec.SkipBytes(8)
+	case ActRecordId:
+		return nil, dec.SkipBytes(76)
+	case ActRecordId2:
+		return nil, dec.SkipBytes(4)
+	case ActSanctumInfinitySwap:
+		// amount u64 + src_lst_value_calc_accs u8 + dst_lst_value_calc_accs u8 + src_lst_index u32 + dst_lst_index u32 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1 + 4 + 4 + 1)
+	case ActSanctumInfinityLiquidity:
+		// amount u64 + lst_value_calc_accs u8 + lst_index u32 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 4 + 1)
+	case ActOpenPredictionsOrder:
+		// nonce u64 + order_outcome u8 + quoted_out_amount u64 + slippage_bps u16 + platform_fee_recipient_vault pubkey(32) + platform_fee_scale u16
+		return nil, dec.SkipBytes(8 + 1 + 8 + 2 + 32 + 2)
+	case ActScorchSwap:
+		// amount u64 + id u128 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 16 + 1)
+	case ActIncludeAccount:
+		return nil, nil
+	default:
+		return nil, fmt.Errorf("unsupported action tag %d", tag)
+	}
+}
+
+// SwapParams: actions Vec<Action>, quoted_out_amount u64, slippage_bps u16, platform_fee_bps u16
+func decodeSwapParams(data []byte) (*dflowSwapParams, error) {
+	dec := bin.NewBorshDecoder(data)
+	out := &dflowSwapParams{}
+	ln, err := dec.ReadUint32(binary.LittleEndian)
+	if err != nil {
+		return nil, err
+	}
+	out.Actions = make([]dflowAction, 0, ln)
+	for i := uint32(0); i < ln; i++ {
+		tag, err := dec.ReadUint8()
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d] tag: %w", i, err)
+		}
+		pump, err := skipDflowAction(dec, tag)
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d]: %w", i, err)
+		}
+		out.Actions = append(out.Actions, dflowAction{Tag: tag, Pump: pump})
+	}
+	return out, nil
+}
+
+// Swap2Params: actions Vec<Action>, quoted_out_amount u64, slippage_bps u16, platform_fee_bps u16, positive_slippage_fee_limit_pct u8
+func decodeSwap2Params(data []byte) (*dflowSwapParams, error) {
+	dec := bin.NewBorshDecoder(data)
+	out := &dflowSwapParams{}
+	ln, err := dec.ReadUint32(binary.LittleEndian)
+	if err != nil {
+		return nil, err
+	}
+	out.Actions = make([]dflowAction, 0, ln)
+	for i := uint32(0); i < ln; i++ {
+		tag, err := dec.ReadUint8()
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d] tag: %w", i, err)
+		}
+		pump, err := skipDflowAction(dec, tag)
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d]: %w", i, err)
+		}
+		out.Actions = append(out.Actions, dflowAction{Tag: tag, Pump: pump})
+	}
+	return out, nil
+}
+
+func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (*TxSignal, error) {
+	msg := tx.Message
+	if instructionIndex >= len(msg.Instructions) {
+		return nil, fmt.Errorf("instruction index out of bounds")
+	}
+	ix := msg.Instructions[instructionIndex]
+	if len(ix.Data) < 8 {
+		return nil, nil
+	}
+
+	var err error
+	disc := ix.Data[:8]
+	payload := ix.Data[8:]
+
+	var params *dflowSwapParams
+	switch {
+	case bytes.Equal(disc, dflowSwapDisc), bytes.Equal(disc, dflowSwapWithDestinationDisc), bytes.Equal(disc, dflowSwapWithDestinationNative):
+		params, err = decodeSwapParams(payload)
+	case bytes.Equal(disc, dflowSwap2Disc), bytes.Equal(disc, dflowSwap2WithDestinationDisc), bytes.Equal(disc, dflowSwap2WithDestinationNative):
+		params, err = decodeSwap2Params(payload)
+	default:
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	if params == nil {
+		return nil, nil
+	}
+
+	var pump *pumpFunAmm
+	for _, act := range params.Actions {
+		if act.Tag == ActPumpFunAmmSell && act.Pump != nil {
+			pump = act.Pump
+			break
+		}
+	}
+	if pump == nil {
+		return nil, nil // only care about PumpFunAmmSell
+	}
+
+	// Require WSOL pair when destination mint is provided.
+	var (
+		srcIdx uint8
+	)
+	for i, acctIdx := range ix.Accounts {
+		if i < 6 {
+			continue
+		}
+		key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
+		if err != nil {
+			return nil, err
+		}
+		if key.Equals(pumpAmmProgramID) {
+			srcIdx = uint8(i + 4)
+			break
+		}
+	}
+	if srcIdx == 0 || srcIdx+1 >= uint8(len(ix.Accounts)) {
+		return nil, nil
+	}
+
+	baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[srcIdx]))
+	if err != nil {
+		return nil, err
+	}
+	quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[srcIdx+1]))
+	if err != nil {
+		return nil, err
+	}
+	if !quoteMint.Equals(solana.WrappedSol) {
+		return nil, nil
+	}
+
+	// Build TxSignal
+	sig := &TxSignal{
+		Label:              "dflow",
+		TxHash:             tx.Signatures[0].String(),
+		Maker:              tx.Message.StaticAccountKeys[0].String(),
+		Program:            "PumpAMM",
+		Event:              "sell",
+		Token0Address:      baseMint.String(),
+		Token1Address:      wsolMint,
+		Token0Amount:       formatTokenAmount(pump.Amount),
+		Token1Amount:       decimal.Zero,
+		Token0AmountUint64: uint64(pump.Amount),
+		Token1AmountUint64: 0,
+	}
+	return sig, nil
+}

pkg/shreder/txparser.go

@@ -278,6 +278,9 @@ func ParseTransaction(update *SubscribeUpdateTransaction, loader *AddressTables)
 		case okxDexRouteV2ProgramID:
 			txRes, err := parseOkxDexRouteV2Instruction(versioned, i)
 			parsed = appendParsed(parsed, txRes, err, txHash, "okxdexroutev2", okxDexRouteV2ProgramID.String())
+		case dflowProgramID:
+			txRes, err := parseDFlowInstruction(versioned, i)
+			parsed = appendParsed(parsed, txRes, err, txHash, "dflow", dflowProgramID.String())
 		}
 	}