Compute moneyFlew from bob log events for transaction status

Fetch all logs for a tick via GET /log/{epoch}/{logIdStart}/{logIdEnd},
group QU_TRANSFER events by transaction hash, and determine moneyFlew
by comparing the transaction fields against its first QU_TRANSFER event:

  moneyFlew = (tx.amount == event.amount) && (tx.src == event.src)
           && (tx.dst == event.dst) && (tx.tick == event.tick)

Only transactions with amount > 0 are candidates; others default to false.

- Add computeMoneyFlew with log fetching and matching logic
- Add transaction caching in BobDataFetcher for use by GetTxStatus
- Add bobLogEvent/bobLogBody types for log deserialization
- Replace GetTxStatus stub with real moneyFlew computation
- Add tests covering matching, mismatched, zero-amount, and empty cases

Author: 0xluk

BOB_GAPS.md

@@ -50,20 +50,21 @@ response.
 
 ---
 
-## 4. Transaction status (moneyFlew)
+## 4. Transaction status (moneyFlew) - RESOLVED
 
 **What**: Bob has per-transaction `executed` status from log events, but does NOT
 expose the `MoneyFlew` bit array that the node connector provides.
 
-**Current behavior**: The archiver returns empty transaction status (all moneyFlew
-bits set to false) when using bob. The `txstatus` validation is effectively skipped.
+**Resolution**: Implemented moneyFlew computation from bob's log events. The archiver
+fetches all logs for a tick via `GET /log/{epoch}/{logIdStart}/{logIdEnd}`, groups
+QU_TRANSFER events by transaction hash, and applies the algorithm:
+```
+moneyFlew = (tx.amount == first_event.amount) && (tx.src == first_event.src)
+         && (tx.dst == first_event.dst) && (tx.tick == first_event.tick)
+```
+Only transactions with `amount > 0` are candidates; others default to false.
 
-**Impact**: Medium. The `moneyFlew` field is used by API consumers to determine if
-a transaction's side effects were executed. Currently all transactions appear as
-"not executed" when using bob.
-
-**Future fix**: Use `qubic_getTransactionReceipt` to fetch per-transaction execution
-status, then reconstruct the MoneyFlew bit array from the `executed` field.
+See: `network/bob/moneyflew.go`
 
 ---
 

network/bob/fetcher.go

@@ -19,6 +19,11 @@ type BobDataFetcher struct {
 	mu             sync.Mutex
 	cachedTick     uint32
 	cachedTickResp *bobTickResponse
+
+	// Per-cycle cache for transactions (from GetTickTransactions).
+	// Used by GetTxStatus to compute moneyFlew without re-fetching.
+	cachedTxsTick uint32
+	cachedTxs     types.Transactions
 }
 
 // NewBobDataFetcher creates a new BobDataFetcher.
@@ -108,6 +113,12 @@ func (b *BobDataFetcher) GetTickTransactions(ctx context.Context, tickNumber uin
 		txs = append(txs, tx)
 	}
 
+	// Cache for use by GetTxStatus
+	b.mu.Lock()
+	b.cachedTxsTick = tickNumber
+	b.cachedTxs = txs
+	b.mu.Unlock()
+
 	return txs, nil
 }
 
@@ -125,17 +136,37 @@ func (b *BobDataFetcher) GetComputors(ctx context.Context) (types.Computors, err
 	return convertComputors(compsResp.Computors, compsResp.Epoch)
 }
 
-func (b *BobDataFetcher) GetTxStatus(_ context.Context, _ uint32) (types.TransactionStatus, bool, error) {
-	// TODO: Compute moneyFlew from bob's log events using qubic_getTransactionReceipt.
-	// For now, return empty status indicating data is not available.
-	return types.TransactionStatus{}, false, nil
+func (b *BobDataFetcher) GetTxStatus(ctx context.Context, tick uint32) (types.TransactionStatus, bool, error) {
+	tickResp, err := b.getTickResponse(ctx, tick)
+	if err != nil {
+		return types.TransactionStatus{}, false, fmt.Errorf("getting tick response for tx status: %w", err)
+	}
+
+	// Get cached transactions (populated by GetTickTransactions which runs before GetTxStatus)
+	b.mu.Lock()
+	txs := b.cachedTxs
+	txsTick := b.cachedTxsTick
+	b.mu.Unlock()
+
+	if txsTick != tick || txs == nil {
+		return types.TransactionStatus{}, false, fmt.Errorf("transactions for tick %d not cached (have tick %d)", tick, txsTick)
+	}
+
+	status, err := computeMoneyFlew(ctx, b.client, tickResp, txs, tick)
+	if err != nil {
+		return types.TransactionStatus{}, false, fmt.Errorf("computing moneyFlew: %w", err)
+	}
+
+	return status, true, nil
 }
 
 func (b *BobDataFetcher) Release(_ error) {
-	// Clear the per-cycle cache
+	// Clear per-cycle caches
 	b.mu.Lock()
 	b.cachedTick = 0
 	b.cachedTickResp = nil
+	b.cachedTxsTick = 0
+	b.cachedTxs = nil
 	b.mu.Unlock()
 }
 

network/bob/moneyflew.go

@@ -0,0 +1,184 @@
+package bob
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"strings"
+
+	"github.com/qubic/go-node-connector/types"
+)
+
+const quTransferLogType = 0
+
+// firstQUTransfer holds the fields from the first QU_TRANSFER event for a transaction.
+type firstQUTransfer struct {
+	Source      string // uppercase identity
+	Destination string // uppercase identity
+	Amount      int64
+	Tick        uint32
+}
+
+// computeMoneyFlew fetches logs for the given tick and computes the moneyFlew
+// status for each transaction by comparing the transaction's fields against its
+// first QU_TRANSFER log event.
+//
+// Algorithm (from bob maintainer):
+//
+//	moneyFlew = (tx.amount == first_event.amount)
+//	         && (tx.src == first_event.src)
+//	         && (tx.dst == first_event.dst)
+//	         && (tx.tick == first_event.tick)
+//
+// Only transactions with amount > 0 are candidates; others default to false.
+func computeMoneyFlew(ctx context.Context, client *Client, tickResp *bobTickResponse, txs types.Transactions, tickNumber uint32) (types.TransactionStatus, error) {
+	epoch := tickResp.TickData.Epoch
+	logIdStart := tickResp.TickData.LogIdStart
+	logIdEnd := tickResp.TickData.LogIdEnd
+
+	// Build transaction lookup: txHash -> Transaction
+	txByHash := make(map[string]*types.Transaction, len(txs))
+	for i := range txs {
+		txID, err := txs[i].ID()
+		if err != nil {
+			return types.TransactionStatus{}, fmt.Errorf("computing tx ID for index %d: %w", i, err)
+		}
+		txByHash[txID] = &txs[i]
+	}
+
+	// Collect non-zero transaction digests and build the digest list
+	var digestList [][32]byte
+	for i, digestStr := range tickResp.TickData.TransactionDigests {
+		if digestStr == "" {
+			continue
+		}
+		digest, err := qubicHashToBytes32(digestStr)
+		if err != nil {
+			return types.TransactionStatus{}, fmt.Errorf("converting digest[%d]: %w", i, err)
+		}
+		digestList = append(digestList, digest)
+	}
+
+	// Build the MoneyFlew bit array
+	var moneyFlew [128]byte
+
+	// If there are no logs, return with all-zero moneyFlew
+	if logIdStart < 0 || logIdEnd < logIdStart {
+		return types.TransactionStatus{
+			CurrentTickOfNode:  tickNumber,
+			Tick:               tickNumber,
+			TxCount:            uint32(len(digestList)),
+			MoneyFlew:          moneyFlew,
+			TransactionDigests: digestList,
+		}, nil
+	}
+
+	// Fetch all logs for this tick
+	firstTransfers, err := fetchFirstQUTransfers(ctx, client, epoch, logIdStart, logIdEnd)
+	if err != nil {
+		return types.TransactionStatus{}, fmt.Errorf("fetching logs for tick %d: %w", tickNumber, err)
+	}
+
+	// For each transaction digest, determine moneyFlew
+	for i, digest := range digestList {
+		// Convert digest to txHash (60-char lowercase)
+		var id types.Identity
+		id, err := id.FromPubKey(digest, true)
+		if err != nil {
+			log.Printf("[WARN] failed to convert digest[%d] to identity: %v", i, err)
+			continue
+		}
+		txHash := id.String()
+
+		// Look up the actual transaction
+		tx, hasTx := txByHash[txHash]
+		if !hasTx || tx.Amount <= 0 {
+			continue // moneyFlew stays false
+		}
+
+		// Look up the first QU_TRANSFER for this tx
+		transfer, hasLog := firstTransfers[txHash]
+		if !hasLog {
+			continue // no QU_TRANSFER event -> moneyFlew = false
+		}
+
+		// Convert tx public keys to uppercase identities for comparison
+		var srcID types.Identity
+		srcID, err = srcID.FromPubKey(tx.SourcePublicKey, false)
+		if err != nil {
+			log.Printf("[WARN] failed to convert source pubkey for tx %s: %v", txHash, err)
+			continue
+		}
+		var dstID types.Identity
+		dstID, err = dstID.FromPubKey(tx.DestinationPublicKey, false)
+		if err != nil {
+			log.Printf("[WARN] failed to convert dest pubkey for tx %s: %v", txHash, err)
+			continue
+		}
+
+		// Apply the moneyFlew algorithm
+		if tx.Amount == transfer.Amount &&
+			string(srcID) == transfer.Source &&
+			string(dstID) == transfer.Destination &&
+			tx.Tick == transfer.Tick {
+			setMoneyFlewBit(&moneyFlew, i)
+		}
+	}
+
+	return types.TransactionStatus{
+		CurrentTickOfNode:  tickNumber,
+		Tick:               tickNumber,
+		TxCount:            uint32(len(digestList)),
+		MoneyFlew:          moneyFlew,
+		TransactionDigests: digestList,
+	}, nil
+}
+
+// fetchFirstQUTransfers fetches all logs for the tick and returns a map of
+// txHash -> first QU_TRANSFER event for that transaction.
+func fetchFirstQUTransfers(ctx context.Context, client *Client, epoch uint16, logIdStart, logIdEnd int64) (map[string]firstQUTransfer, error) {
+	path := fmt.Sprintf("/log/%d/%d/%d", epoch, logIdStart, logIdEnd)
+	body, err := client.RESTGet(ctx, path)
+	if err != nil {
+		return nil, fmt.Errorf("GET %s: %w", path, err)
+	}
+
+	var logs []bobLogEvent
+	if err := json.Unmarshal(body, &logs); err != nil {
+		return nil, fmt.Errorf("unmarshalling logs: %w", err)
+	}
+
+	result := make(map[string]firstQUTransfer)
+	for _, logEvent := range logs {
+		if !logEvent.OK {
+			continue
+		}
+		if logEvent.Type != quTransferLogType {
+			continue
+		}
+		txHash := strings.ToLower(logEvent.TxHash)
+		if txHash == "" {
+			continue
+		}
+		// Only keep the first QU_TRANSFER per transaction
+		if _, exists := result[txHash]; exists {
+			continue
+		}
+		result[txHash] = firstQUTransfer{
+			Source:      logEvent.Body.From,
+			Destination: logEvent.Body.To,
+			Amount:      logEvent.Body.Amount,
+			Tick:        logEvent.Tick,
+		}
+	}
+
+	return result, nil
+}
+
+// setMoneyFlewBit sets bit at position index in the MoneyFlew byte array.
+func setMoneyFlewBit(moneyFlew *[128]byte, index int) {
+	bytePos := index / 8
+	bitPos := index % 8
+	moneyFlew[bytePos] |= 1 << bitPos
+}