db+waddrmgr: add scan-batch scaffolding and rollback support

wallet/internal/bwtest/mock/store.go

@@ -380,6 +380,18 @@ func (m *Store) DeleteExpiredLeases(ctx context.Context,
 	return args.Error(0)
 }
 
+// ListOutputsToWatch implements the db.UTXOStore interface.
+func (m *Store) ListOutputsToWatch(ctx context.Context,
+	walletID uint32) ([]db.UtxoInfo, error) {
+
+	args := m.Called(ctx, walletID)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).([]db.UtxoInfo), args.Error(1)
+}
+
 // Balance implements the db.UTXOStore interface.
 func (m *Store) Balance(ctx context.Context,
 	params db.BalanceParams) (db.BalanceResult, error) {
@@ -410,6 +422,15 @@ func (m *Store) UpdateTx(ctx context.Context,
 	return args.Error(0)
 }
 
+// ApplyTxBatch implements the db.TxStore interface.
+func (m *Store) ApplyTxBatch(ctx context.Context,
+	params db.TxBatchParams) error {
+
+	args := m.Called(ctx, params)
+
+	return args.Error(0)
+}
+
 // GetTx implements the db.TxStore interface.
 func (m *Store) GetTx(ctx context.Context,
 	query db.GetTxQuery) (*db.TxInfo, error) {

wallet/internal/db/data_types.go

@@ -1078,6 +1078,20 @@ type CreateTxParams struct {
 	Credits map[uint32]address.Address
 }
 
+// TxBatchParams contains a wallet transaction batch and optional sync-tip
+// update that should be applied atomically.
+type TxBatchParams struct {
+	// WalletID is the ID of the wallet receiving the batch.
+	WalletID uint32
+
+	// Transactions contains the transaction records to apply.
+	Transactions []CreateTxParams
+
+	// SyncedTo optionally records the wallet's new chain sync tip as part of
+	// the same batch.
+	SyncedTo *Block
+}
+
 // UpdateTxState contains one requested transaction-state change.
 type UpdateTxState struct {
 	// Block records the transaction as confirmed in the provided block.

wallet/internal/db/interface.go

@@ -411,6 +411,10 @@ type TxStore interface {
 	// graph-affecting lifecycle changes.
 	UpdateTx(ctx context.Context, params UpdateTxParams) error
 
+	// ApplyTxBatch atomically records a batch of transaction records and an
+	// optional wallet sync-tip update.
+	ApplyTxBatch(ctx context.Context, params TxBatchParams) error
+
 	// GetTx retrieves a transaction record by its hash. It takes a context
 	// and GetTxQuery, returning a TxInfo struct or an error if the
 	// transaction is not found.
@@ -517,6 +521,10 @@ type UTXOStore interface {
 	// DeleteExpiredLeases removes expired UTXO lease records for the wallet.
 	DeleteExpiredLeases(ctx context.Context, walletID uint32) error
 
+	// ListOutputsToWatch returns UTXOs that recovery scans should watch.
+	ListOutputsToWatch(ctx context.Context, walletID uint32) ([]UtxoInfo,
+		error)
+
 	// Balance returns a wallet-scoped balance view for the current unspent UTXO
 	// set after applying any optional caller-supplied filters.
 	//

wallet/internal/db/itest/pg_test.go

@@ -408,3 +408,29 @@ func walletUtxoExists(t *testing.T, store *pg.Store,
 
 	return true
 }
+
+// walletUtxoSpent reports whether one wallet-scoped outpoint exists and is
+// recorded as spent, i.e. its spend edge points at a spending transaction.
+func walletUtxoSpent(t *testing.T, store *pg.Store,
+	walletID uint32,
+	outPoint wire.OutPoint) bool {
+
+	t.Helper()
+
+	spentBy, err := store.Queries().GetUtxoSpendByOutpoint(
+		t.Context(), sqlc.GetUtxoSpendByOutpointParams{
+			WalletID:    int64(walletID),
+			TxHash:      outPoint.Hash[:],
+			OutputIndex: int32(outPoint.Index),
+		},
+	)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return false
+		}
+
+		require.NoError(t, err)
+	}
+
+	return spentBy.SpentByTxID.Valid
+}

wallet/internal/db/itest/sqlite_test.go

@@ -153,3 +153,29 @@ func walletUtxoExists(t *testing.T, store *sqlite.Store,
 
 	return true
 }
+
+// walletUtxoSpent reports whether one wallet-scoped outpoint exists and is
+// recorded as spent, i.e. its spend edge points at a spending transaction.
+func walletUtxoSpent(t *testing.T, store *sqlite.Store,
+	walletID uint32,
+	outPoint wire.OutPoint) bool {
+
+	t.Helper()
+
+	spentBy, err := store.Queries().GetUtxoSpendByOutpoint(
+		t.Context(), sqlc.GetUtxoSpendByOutpointParams{
+			WalletID:    int64(walletID),
+			TxHash:      outPoint.Hash[:],
+			OutputIndex: int64(outPoint.Index),
+		},
+	)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return false
+		}
+
+		require.NoError(t, err)
+	}
+
+	return spentBy.SpentByTxID.Valid
+}

wallet/internal/db/itest/txstore_create_edge_cases_test.go

@@ -66,9 +66,9 @@ func TestCreateTxRejectsInvalidParams(t *testing.T) {
 	require.ErrorContains(t, err, "tx is required")
 }
 
-// TestCreateTxRejectsDuplicateConfirmedTransaction verifies that duplicate
-// confirmed inserts fail instead of silently creating a second row.
-func TestCreateTxRejectsDuplicateConfirmedTransaction(t *testing.T) {
+// TestCreateTxSkipsDuplicateConfirmedTransaction verifies that an exact
+// duplicate confirmed insert is idempotent instead of creating a second row.
+func TestCreateTxSkipsDuplicateConfirmedTransaction(t *testing.T) {
 	t.Parallel()
 
 	store := NewTestStore(t)
@@ -98,7 +98,7 @@ func TestCreateTxRejectsDuplicateConfirmedTransaction(t *testing.T) {
 	require.NoError(t, err)
 
 	err = store.CreateTx(t.Context(), params)
-	require.ErrorIs(t, err, db.ErrTxAlreadyExists)
+	require.NoError(t, err)
 }
 
 // TestCreateTxRejectsMissingConfirmingBlockForExistingUnminedRow verifies that