Add dflow pumpfun parse

pkg/shreder/dflow.go

@@ -86,28 +86,28 @@ const (
 	drv1Nexus
 )
 
-// PumpFunAmmSellOptions { amount: u64, orchestrator_flags: OrchestratorFlags{flags u8} }
+// PumpFun*Options { amount: u64, orchestrator_flags: OrchestratorFlags{flags u8} }
-type pumpFunAmm struct {
+type pumpFunAction struct {
 	Amount uint64
 	Flags  uint8
 }
 
 type dflowAction struct {
 	Tag  uint8
-	Pump *pumpFunAmm
+	Pump *pumpFunAction
 }
 
 type dflowSwapParams struct {
 	Actions []dflowAction
 }
 
-// bytes to skip for Action variants before/after PumpFunAmmSell; only PumpFunAmmSell is decoded.
+// bytes to skip for Action variants; only PumpFun* actions are decoded.
-func skipDflowAction(dec *bin.Decoder, tag uint8) (*pumpFunAmm, error) {
+func skipDflowAction(dec *bin.Decoder, tag uint8) (*pumpFunAction, 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,
+	case ActRaydiumAmmSwap, ActLifinityV2Swap, ActObricV2Swap,
 		ActSolFiSwap, ActRubiconSwap, ActMeteoraDammV1Swap, ActRaydiumCpSwap,
 		ActStabbleStableSwap, ActTesseraVSwap, ActMeteoraDammV2Swap, ActRaydiumLaunchlabSwap,
 		ActZeroFiSwap, ActAlphaQSwap, ActTokenSwap, ActSolFiV2Swap, ActMozartSwap, ActHeavenSwap,
@@ -123,7 +123,7 @@ func skipDflowAction(dec *bin.Decoder, tag uint8) (*pumpFunAmm, error) {
 	case ActGammaSwap:
 		// amount u64 + endorsed bool + orchestrator_flags u8
 		return nil, dec.SkipBytes(8 + 1 + 1)
-	case ActPumpFunAmmSell, ActPumpFunAmmBuy:
+	case ActPumpFunAmmSell, ActPumpFunAmmBuy, ActPumpFunBuy, ActPumpFunSell:
 		amt, err := dec.ReadUint64(binary.LittleEndian)
 		if err != nil {
 			return nil, err
@@ -132,7 +132,7 @@ func skipDflowAction(dec *bin.Decoder, tag uint8) (*pumpFunAmm, error) {
 		if err != nil {
 			return nil, err
 		}
-		return &pumpFunAmm{Amount: amt, Flags: flg}, nil
+		return &pumpFunAction{Amount: amt, Flags: flg}, nil
 	case ActMeteoraDbcSwap:
 		// amount u64 + is_rate_limiter_applied bool + orchestrator_flags u8
 		return nil, dec.SkipBytes(8 + 1 + 1)
@@ -232,7 +232,34 @@ func decodeSwap2Params(data []byte) (*dflowSwapParams, error) {
 	return out, nil
 }
 
-func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (*TxSignal, error) {
+func findDflowPumpAmmMints(staticKeys []solana.PublicKey, accounts []uint8) (solana.PublicKey, solana.PublicKey, bool, error) {
+	for i, acctIdx := range accounts {
+		key, err := getStaticKey(staticKeys, int(acctIdx))
+		if err != nil {
+			return solana.PublicKey{}, solana.PublicKey{}, false, err
+		}
+		if !key.Equals(pumpAmmProgramID) {
+			continue
+		}
+		baseIdx := i + 4
+		quoteIdx := i + 5
+		if baseIdx >= len(accounts) || quoteIdx >= len(accounts) {
+			return solana.PublicKey{}, solana.PublicKey{}, false, nil
+		}
+		baseMint, err := getStaticKey(staticKeys, int(accounts[baseIdx]))
+		if err != nil {
+			return solana.PublicKey{}, solana.PublicKey{}, false, err
+		}
+		quoteMint, err := getStaticKey(staticKeys, int(accounts[quoteIdx]))
+		if err != nil {
+			return solana.PublicKey{}, solana.PublicKey{}, false, err
+		}
+		return baseMint, quoteMint, true, nil
+	}
+	return solana.PublicKey{}, solana.PublicKey{}, false, nil
+}
+
+func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (TxSignalBatch, error) {
 	msg := tx.Message
 	if instructionIndex >= len(msg.Instructions) {
 		return nil, fmt.Errorf("instruction index out of bounds")
@@ -262,63 +289,121 @@ func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (*TxS
 		return nil, nil
 	}
 
-	var pump *pumpFunAmm
+	var (
+		pumpAmmBuy  *pumpFunAction
+		pumpAmmSell *pumpFunAction
+		pumpBuy     *pumpFunAction
+		pumpSell    *pumpFunAction
+	)
 	for _, act := range params.Actions {
-		if act.Tag == ActPumpFunAmmSell && act.Pump != nil {
+		if act.Pump == nil {
-			pump = act.Pump
+			continue
-			break
+		}
+		switch act.Tag {
+		case ActPumpFunAmmSell:
+			pumpAmmSell = act.Pump
+		case ActPumpFunAmmBuy:
+			pumpAmmBuy = act.Pump
+		case ActPumpFunBuy:
+			pumpBuy = act.Pump
+		case ActPumpFunSell:
+			pumpSell = act.Pump
 		}
 	}
-	if pump == nil {
-		return nil, nil // only care about PumpFunAmmSell
-	}
 
-	// Require WSOL pair when destination mint is provided.
+	out := make(TxSignalBatch, 0, 2)
-	var (
+	if pumpAmmSell != nil || pumpAmmBuy != nil {
-		srcIdx uint8
+		event := "sell"
-	)
+		amt := pumpAmmSell
-	if len(ix.Accounts) <= 6 {
+		isBuy := false
-		return nil, nil
+		if amt == nil {
-	}
+			event = "buy"
-	accounts := ix.Accounts[5:]
+			isBuy = true
-	for i, acctIdx := range accounts {
+			amt = pumpAmmBuy
-		key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
+		}
+		baseMint, quoteMint, ok, err := findDflowPumpAmmMints(tx.Message.StaticAccountKeys, ix.Accounts)
 		if err != nil {
 			return nil, err
 		}
-		if key.Equals(pumpAmmProgramID) {
+		if ok && quoteMint.Equals(solana.WrappedSol) {
-			srcIdx = uint8(i + 4)
+			var (
-			break
+				token0Amount       decimal.Decimal
+				token1Amount       decimal.Decimal
+				token0AmountUint64 uint64
+				token1AmountUint64 uint64
+				exactSol           bool
+			)
+			if isBuy {
+				exactSol = true
+				token1Amount = formatSolAmount(amt.Amount)
+				token1AmountUint64 = amt.Amount
+			} else {
+				token0Amount = formatTokenAmount(amt.Amount)
+				token0AmountUint64 = amt.Amount
+			}
+			out = append(out, &TxSignal{
+				TxHash:             tx.Signatures[0].String(),
+				Maker:              tx.Message.StaticAccountKeys[0].String(),
+				Program:            "PumpAMM",
+				Event:              event,
+				Token0Address:      baseMint.String(),
+				Token1Address:      wsolMint,
+				Token0Amount:       token0Amount,
+				Token1Amount:       token1Amount,
+				ExactSOL:           exactSol,
+				Token0AmountUint64: token0AmountUint64,
+				Token1AmountUint64: token1AmountUint64,
+			})
 		}
 	}
-	if srcIdx == 0 || srcIdx+1 >= uint8(len(accounts)) {
+
-		return nil, nil
+	if pumpSell != nil || pumpBuy != nil {
+		event := "sell"
+		amt := pumpSell
+		isBuy := false
+		if amt == nil {
+			event = "buy"
+			isBuy = true
+			amt = pumpBuy
+		}
+		mint, ok, err := findPumpFunMint(tx.Message.StaticAccountKeys, ix.Accounts)
+		if err != nil {
+			return nil, err
+		}
+		if ok {
+			var (
+				token0Amount       decimal.Decimal
+				token1Amount       decimal.Decimal
+				token0AmountUint64 uint64
+				token1AmountUint64 uint64
+				exactSol           bool
+			)
+			if isBuy {
+				exactSol = true
+				token1Amount = formatSolAmount(amt.Amount)
+				token1AmountUint64 = amt.Amount
+			} else {
+				token0Amount = formatTokenAmount(amt.Amount)
+				token0AmountUint64 = amt.Amount
+			}
+			out = append(out, &TxSignal{
+				TxHash:             tx.Signatures[0].String(),
+				Maker:              tx.Message.StaticAccountKeys[0].String(),
+				Program:            "Pump",
+				Event:              event,
+				Token0Address:      mint.String(),
+				Token1Address:      wsolMint,
+				Token0Amount:       token0Amount,
+				Token1Amount:       token1Amount,
+				ExactSOL:           exactSol,
+				Token0AmountUint64: token0AmountUint64,
+				Token1AmountUint64: token1AmountUint64,
+			})
+		}
 	}
 
-	baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
+	if len(out) == 0 {
-	if err != nil {
-		return nil, err
-	}
-	quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
-	if err != nil {
-		return nil, err
-	}
-	if !quoteMint.Equals(solana.WrappedSol) {
 		return nil, nil
 	}
-
+	return out, nil
-	// Build TxSignal
-	sig := &TxSignal{
-		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

@@ -360,7 +360,7 @@ func ParseTransaction(update *SubscribeUpdateTransaction, loader *AddressTables,
 			parsed = appendParsed(now, parsed, txRes, err, txHash, "okxdexroutev2")
 		case dflowProgramID:
 			txRes, err := parseDFlowInstruction(versioned, i)
-			parsed = appendParsed(now, parsed, txRes, err, txHash, "dflow")
+			parsed = appendParsedBatch(now, parsed, txRes, err, txHash, "dflow")
 		case gmgnProgramID:
 			txRes, err := parseGMGNInstruction(versioned, i)
 			parsed = appendParsed(now, parsed, txRes, err, txHash, "gmgn")
@@ -397,6 +397,34 @@ func appendParsed(start time.Time, list []*TxSignal, parsed *TxSignal, err error
 	return list
 }
 
+func appendParsedBatch(start time.Time, list []*TxSignal, parsed []*TxSignal, err error, txHash [64]byte, label string) []*TxSignal {
+	if err != nil {
+		if !strings.HasPrefix(err.Error(), "account index") {
+			logger.Debug("txparser: failed to parse", "label", label, "instruction", err, "tx_hash", base58.Encode(txHash[:]))
+		}
+		return list
+	}
+	if len(parsed) == 0 {
+		return list
+	}
+	var end time.Time
+	if !start.IsZero() {
+		end = time.Now()
+	}
+	for _, sig := range parsed {
+		if sig == nil {
+			continue
+		}
+		sig.Label = label
+		if !start.IsZero() {
+			sig.ParseEnd = end
+			sig.ParseStart = start
+		}
+		list = append(list, sig)
+	}
+	return list
+}
+
 func toVersionedTransaction(update *SubscribeUpdateTransaction) (*versionedTransaction, error) {
 	if update == nil || update.Transaction == nil || update.Transaction.Message == nil {
 		return nil, fmt.Errorf("transaction is nil")