wallet: route relevant tx notifications

Send relevant transaction notifications through db.Store so the
dispatcher no longer reads or writes directly through legacy
txstore paths.

Author: yyforyongyu

wallet/syncer.go

@@ -657,6 +657,79 @@ func (s *syncer) updateSyncTip(ctx context.Context,
 	return nil
 }
 
+// putTxNotifications records relevant transaction notifications through the
+// store when configured, falling back to the legacy walletdb path otherwise.
+func (s *syncer) putTxNotifications(ctx context.Context,
+	matches TxEntries, blockMeta *wtxmgr.BlockMeta) error {
+
+	if s.store == nil {
+		return s.DBPutTxns(ctx, matches, blockMeta)
+	}
+
+	var block *db.Block
+	if blockMeta != nil {
+		var err error
+
+		block, err = storeBlockFromBlockMeta(*blockMeta)
+		if err != nil {
+			return err
+		}
+	}
+
+	transactions := make([]db.CreateTxParams, 0, len(matches))
+	for i := range matches {
+		match := matches[i]
+		credits := make(map[uint32]btcutil.Address, len(match.Entries))
+
+		for _, entry := range match.Entries {
+			index := entry.Credit.Index
+			if uint64(index) >= uint64(len(match.Rec.MsgTx.TxOut)) {
+				return fmt.Errorf("credit output %d: %w", index,
+					db.ErrInvalidParam)
+			}
+
+			pkScript := match.Rec.MsgTx.TxOut[index].PkScript
+			_, err := s.store.GetAddress(
+				ctx, db.GetAddressQuery{
+					WalletID:     s.walletID,
+					ScriptPubKey: pkScript,
+				},
+			)
+			if errors.Is(err, db.ErrAddressNotFound) {
+				continue
+			}
+
+			if err != nil {
+				return fmt.Errorf("resolve tx credit %d: %w", index,
+					err)
+			}
+
+			credits[index] = entry.Address
+		}
+
+		transactions = append(transactions, db.CreateTxParams{
+			WalletID: s.walletID,
+			Tx:       &match.Rec.MsgTx,
+			Received: match.Rec.Received,
+			Block:    block,
+			Status:   db.TxStatusPublished,
+			Credits:  credits,
+		})
+	}
+
+	err := s.store.ApplyTxBatch(
+		ctx, db.TxBatchParams{
+			WalletID:     s.walletID,
+			Transactions: transactions,
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("apply tx notifications: %w", err)
+	}
+
+	return nil
+}
+
 // storeBlockFromBlockMeta converts chain notification block metadata into the
 // store block shape.
 func storeBlockFromBlockMeta(block wtxmgr.BlockMeta) (*db.Block, error) {
@@ -1287,7 +1360,7 @@ func (s *syncer) processChainUpdate(ctx context.Context, update any) error {
 	// handled atomically via FilteredBlockConnected.
 	case chain.RelevantTx:
 		matches := s.prepareTxMatches([]*wtxmgr.TxRecord{n.TxRecord})
-		return s.DBPutTxns(ctx, matches, n.Block)
+		return s.putTxNotifications(ctx, matches, n.Block)
 
 	case chain.FilteredBlockConnected:
 		matches := s.prepareTxMatches(n.RelevantTxs)

wallet/syncer_test.go

@@ -831,6 +831,104 @@ func TestHandleChainUpdate(t *testing.T) {
 	require.NoError(t, err)
 }
 
+// matchGetAddressByScript returns a matcher asserting a GetAddress query
+// targets the given wallet and script.
+func matchGetAddressByScript(walletID uint32, pkScript []byte) any {
+	return mock.MatchedBy(func(query db.GetAddressQuery) bool {
+		return query.WalletID == walletID &&
+			bytes.Equal(query.ScriptPubKey, pkScript)
+	})
+}
+
+// matchRelevantTxParams reports whether the single batched transaction carries
+// the expected wallet, hash, receive time, credit address, and published
+// status. Block confirmation is validated separately by the caller.
+func matchRelevantTxParams(txParams db.CreateTxParams, walletID uint32,
+	tx *wire.MsgTx, addr btcutil.Address, received time.Time) bool {
+
+	creditAddr, ok := txParams.Credits[0]
+	if !ok || creditAddr == nil || txParams.Tx == nil {
+		return false
+	}
+
+	return txParams.WalletID == walletID &&
+		txParams.Tx.TxHash() == tx.TxHash() &&
+		txParams.Received.Equal(received) &&
+		txParams.Status == db.TxStatusPublished &&
+		creditAddr.EncodeAddress() == addr.EncodeAddress()
+}
+
+// matchUnminedTxBatch returns a matcher asserting an unmined relevant
+// transaction is written as a single store batch with the expected credit and
+// no confirming block.
+func matchUnminedTxBatch(walletID uint32, tx *wire.MsgTx,
+	addr btcutil.Address, received time.Time) any {
+
+	return mock.MatchedBy(func(params db.TxBatchParams) bool {
+		if params.WalletID != walletID ||
+			len(params.Transactions) != 1 {
+
+			return false
+		}
+
+		txParams := params.Transactions[0]
+		if txParams.Block != nil {
+			return false
+		}
+
+		return matchRelevantTxParams(
+			txParams, walletID, tx, addr, received,
+		)
+	})
+}
+
+// TestProcessRelevantTxUsesStore verifies that relevant transaction
+// notifications are routed through the store when store wiring is available.
+func TestProcessRelevantTxUsesStore(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Create a syncer with store wiring and a transaction paying to a
+	// wallet-owned address.
+	const walletID uint32 = 7
+
+	store := &walletmock.Store{}
+	publisher := &mockTxPublisher{}
+	s := newSyncer(
+		Config{ChainParams: &chainParams}, nil, nil, publisher,
+		syncerStoreConfig{store: store, walletID: walletID},
+	)
+
+	addr, err := btcutil.NewAddressPubKeyHash(
+		make([]byte, 20), &chainParams,
+	)
+	require.NoError(t, err)
+
+	pkScript, err := txscript.PayToAddrScript(addr)
+	require.NoError(t, err)
+
+	tx := wire.NewMsgTx(1)
+	tx.AddTxOut(&wire.TxOut{Value: 1000, PkScript: pkScript})
+
+	received := time.Unix(123, 0).UTC()
+	rec, err := wtxmgr.NewTxRecordFromMsgTx(tx, received)
+	require.NoError(t, err)
+
+	store.On("GetAddress", mock.Anything,
+		matchGetAddressByScript(walletID, pkScript),
+	).Return(&db.AddressInfo{ScriptPubKey: pkScript}, nil).Once()
+
+	store.On("ApplyTxBatch", mock.Anything,
+		matchUnminedTxBatch(walletID, tx, addr, received),
+	).Return(nil).Once()
+
+	// Act: Process an unconfirmed relevant transaction notification.
+	err = s.processChainUpdate(t.Context(), chain.RelevantTx{TxRecord: rec})
+
+	// Assert: The notification was written through the store batch API.
+	require.NoError(t, err)
+	store.AssertExpectations(t)
+}
+
 // TestExtractAddrEntries verifies address extraction from outputs.
 func TestExtractAddrEntries(t *testing.T) {
 	t.Parallel()