pg: use native pgx store queries

Generate PostgreSQL queries for pgx/v5, then run normal reads through
pgxpool and writes through pgx.Tx. Adapt generated query values and row
tests while retaining database/sql solely for migrations and tagged test
administration.

Author: yyforyongyu

go.mod

@@ -25,7 +25,6 @@ require (
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/jessevdk/go-flags v1.6.1
 	github.com/jrick/logrotate v1.1.2
-	github.com/lib/pq v1.12.3
 	github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf
 	github.com/lightninglabs/neutrino v0.18.0
 	github.com/lightninglabs/neutrino/cache v1.1.4
@@ -78,6 +77,7 @@ require (
 	github.com/kcalvinalvin/anet v0.0.0-20251112173137-d8ddc1f6dbee // indirect
 	github.com/kkdai/bstream v1.0.0 // indirect
 	github.com/klauspost/compress v1.18.5 // indirect
+	github.com/lib/pq v1.12.3 // indirect
 	github.com/lightningnetwork/lnd/clock v1.0.1 // indirect
 	github.com/lightningnetwork/lnd/queue v1.0.1 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect

sqlc.yaml

@@ -11,7 +11,7 @@ sql:
         package: "sqlc"
 
         # This is the driver package that sqlc will use in the generated code.
-        sql_package: database/sql
+        sql_package: pgx/v5
 
         # Generate a `Querier` interface of all query methods. It's useful for
         # mocking in tests.
@@ -20,7 +20,7 @@ sql:
         # Export generated SQL statements so they're usable from other packages.
         emit_exported_queries: true
 
-        # Generate prepared statements for better performance with database/sql.
+        # Generate prepared statements for better performance with pgx.
         emit_prepared_queries: true
 
   - engine: "sqlite"

wallet/internal/db/itest/fixtures_pg_test.go

@@ -13,6 +13,7 @@ import (
 	dberr "github.com/btcsuite/btcwallet/wallet/internal/db/err"
 	"github.com/btcsuite/btcwallet/wallet/internal/sql/pg/sqlc"
 	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/stretchr/testify/require"
 )
 
@@ -79,12 +80,12 @@ func CreateAccountWithNumber(t *testing.T, queries *sqlc.Queries,
 		t.Context(), sqlc.CreateDerivedAccountParams{
 			ScopeID:     scopeID,
 			AccountName: name,
-			AccountNumber: sql.NullInt64{
+			AccountNumber: pgtype.Int8{
 				Int64: int64(accountNumber),
 				Valid: true,
 			},
 			PublicKey:         RandomBytes(32),
-			MasterFingerprint: sql.NullInt64{},
+			MasterFingerprint: pgtype.Int8{},
 		},
 	)
 	require.NoError(t, err)

wallet/internal/db/itest/pg_test.go

@@ -23,14 +23,61 @@ import (
 	"github.com/btcsuite/btcd/wire/v2"
 	"github.com/btcsuite/btcwallet/wallet/internal/db"
 	"github.com/btcsuite/btcwallet/wallet/internal/db/pg"
+	pgschema "github.com/btcsuite/btcwallet/wallet/internal/sql/pg"
 	"github.com/btcsuite/btcwallet/wallet/internal/sql/pg/sqlc"
 	"github.com/docker/go-connections/nat"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/stdlib"
 	"github.com/stretchr/testify/require"
 	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/modules/postgres"
 	"github.com/testcontainers/testcontainers-go/wait"
 )
 
+// postgresTestStore owns integration-only adapters around a PostgreSQL Store.
+type postgresTestStore struct {
+	*pg.Store
+
+	sqlDB *sql.DB
+}
+
+var _ db.Store = (*postgresTestStore)(nil)
+
+// DB returns the shared-pool database/sql adapter used by integration tests.
+func (s *postgresTestStore) DB() *sql.DB {
+	return s.sqlDB
+}
+
+// RollbackAllMigrations rolls back all PostgreSQL migrations.
+func (s *postgresTestStore) RollbackAllMigrations() error {
+	return pgschema.RollbackMigrations(context.Background(), s.sqlDB)
+}
+
+// ApplyAllMigrations reapplies all PostgreSQL migrations.
+func (s *postgresTestStore) ApplyAllMigrations() error {
+	return pgschema.ApplyMigrations(context.Background(), s.sqlDB)
+}
+
+// Close closes the integration adapter and native PostgreSQL Store.
+func (s *postgresTestStore) Close() error {
+	err := s.sqlDB.Close()
+	storeErr := s.Store.Close()
+
+	if err != nil {
+		return fmt.Errorf("close integration database: %w", err)
+	}
+
+	return storeErr
+}
+
+// isPostgresTestStore reports whether store is the PostgreSQL test backend.
+func isPostgresTestStore(store any) bool {
+	_, ok := store.(*postgresTestStore)
+
+	return ok
+}
+
 const (
 	// pgMaxIdentifierLen is the PostgreSQL maximum identifier length
 	// (NAMEDATALEN - 1).
@@ -249,7 +296,7 @@ func sanitizedPgDBName(t *testing.T) string {
 // limit allows, exhausting the PostgreSQL connection pool. Avoid this by
 // creating NewTestStore inside each parallel subtest so its lifecycle is tied
 // to the subtest's parallel slot.
-func NewTestStore(t *testing.T) *pg.Store {
+func NewTestStore(t *testing.T) *postgresTestStore {
 	t.Helper()
 
 	return NewTestStoreWithDerive(t, mockDeriveFunc())
@@ -258,7 +305,16 @@ func NewTestStore(t *testing.T) *pg.Store {
 // NewTestStoreWithDerive creates a new PostgreSQL database for testing with the
 // provided address derivation function.
 func NewTestStoreWithDerive(t *testing.T,
-	deriveAddress db.AddressDerivationFunc) *pg.Store {
+	deriveAddress db.AddressDerivationFunc) *postgresTestStore {
+
+	t.Helper()
+
+	return newTestStore(t, deriveAddress, 0)
+}
+
+// newTestStore creates a PostgreSQL test store with an explicit pool limit.
+func newTestStore(t *testing.T, deriveAddress db.AddressDerivationFunc,
+	maxConnections int) *postgresTestStore {
 
 	t.Helper()
 	ctx := t.Context()
@@ -291,23 +347,52 @@ func NewTestStoreWithDerive(t *testing.T,
 
 	cfg := pg.Config{
 		Dsn:            testConnStr,
-		MaxConnections: 0,
+		MaxConnections: maxConnections,
 		DeriveAddress:  deriveAddress,
 	}
 
 	store, err := pg.NewStore(t.Context(), cfg)
 	require.NoError(t, err, "failed to create postgres store")
 
+	testStore := &postgresTestStore{
+		Store: store,
+		sqlDB: stdlib.OpenDBFromPool(store.Pool()),
+	}
+
 	t.Cleanup(func() {
-		_ = store.Close()
+		_ = testStore.Close()
 	})
 
-	return store
+	return testStore
+}
+
+// TestPostgresTestStoreSharesPool verifies the integration adapter releases
+// its connection back to the native pool when only one pool slot is available.
+func TestPostgresTestStoreSharesPool(t *testing.T) {
+	t.Parallel()
+
+	store := newTestStore(t, mockDeriveFunc(), 1)
+	sqlDB := store.DB()
+	require.Same(t, sqlDB, store.DB())
+
+	ctx, cancel := context.WithTimeout(
+		t.Context(), db.DefaultConnectionTimeout,
+	)
+	defer cancel()
+
+	require.NoError(t, sqlDB.PingContext(ctx))
+	require.Zero(t, store.Pool().Stat().AcquiredConns())
+
+	wallets, err := store.Queries().ListWallets(
+		ctx, sqlc.ListWalletsParams{PageLimit: 1},
+	)
+	require.NoError(t, err)
+	require.Empty(t, wallets)
 }
 
 // childSpendingTxIDs returns the direct child transaction IDs recorded for the
 // provided parent transaction hash.
-func childSpendingTxIDs(t *testing.T, store *pg.Store,
+func childSpendingTxIDs(t *testing.T, store *postgresTestStore,
 	walletID uint32,
 	txHash chainhash.Hash) []int64 {
 
@@ -340,7 +425,7 @@ func childSpendingTxIDs(t *testing.T, store *pg.Store,
 
 // txIDByHash returns the database row ID for the given wallet-scoped
 // transaction hash and reports whether the row exists.
-func txIDByHash(t *testing.T, store *pg.Store, walletID uint32,
+func txIDByHash(t *testing.T, store *postgresTestStore, walletID uint32,
 	txHash chainhash.Hash) (int64, bool) {
 
 	t.Helper()
@@ -352,7 +437,7 @@ func txIDByHash(t *testing.T, store *pg.Store, walletID uint32,
 		},
 	)
 	if err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
+		if errors.Is(err, pgx.ErrNoRows) {
 			return 0, false
 		}
 
@@ -364,7 +449,7 @@ func txIDByHash(t *testing.T, store *pg.Store, walletID uint32,
 
 // setTxStatus rewrites one wallet-scoped transaction row to the provided
 // status using the internal status-update query.
-func setTxStatus(t *testing.T, store *pg.Store, walletID uint32,
+func setTxStatus(t *testing.T, store *postgresTestStore, walletID uint32,
 	txHash chainhash.Hash, status db.TxStatus) {
 
 	t.Helper()
@@ -385,7 +470,7 @@ func setTxStatus(t *testing.T, store *pg.Store, walletID uint32,
 
 // walletUtxoExists reports whether one wallet-scoped outpoint is currently
 // present in the UTXO set.
-func walletUtxoExists(t *testing.T, store *pg.Store,
+func walletUtxoExists(t *testing.T, store *postgresTestStore,
 	walletID uint32,
 	outPoint wire.OutPoint) bool {
 
@@ -399,7 +484,7 @@ func walletUtxoExists(t *testing.T, store *pg.Store,
 		},
 	)
 	if err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
+		if errors.Is(err, pgx.ErrNoRows) {
 			return false
 		}
 
@@ -411,7 +496,7 @@ func walletUtxoExists(t *testing.T, store *pg.Store,
 
 // 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,
+func walletUtxoSpent(t *testing.T, store *postgresTestStore,
 	walletID uint32,
 	outPoint wire.OutPoint) bool {
 
@@ -425,7 +510,7 @@ func walletUtxoSpent(t *testing.T, store *pg.Store,
 		},
 	)
 	if err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
+		if errors.Is(err, pgx.ErrNoRows) {
 			return false
 		}
 
@@ -436,7 +521,7 @@ func walletUtxoSpent(t *testing.T, store *pg.Store,
 }
 
 // clearUtxosSpentByTxID clears all UTXO spend edges claimed by one transaction.
-func clearUtxosSpentByTxID(t *testing.T, store *pg.Store,
+func clearUtxosSpentByTxID(t *testing.T, store *postgresTestStore,
 	walletID uint32, txHash chainhash.Hash) {
 
 	t.Helper()
@@ -447,7 +532,7 @@ func clearUtxosSpentByTxID(t *testing.T, store *pg.Store,
 	rows, err := store.Queries().ClearUtxosSpentByTxID(
 		t.Context(), sqlc.ClearUtxosSpentByTxIDParams{
 			WalletID: int64(walletID),
-			SpentByTxID: sql.NullInt64{
+			SpentByTxID: pgtype.Int8{
 				Int64: txID,
 				Valid: true,
 			},

wallet/internal/db/itest/sqlite_test.go

@@ -16,6 +16,11 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// isPostgresTestStore reports whether store is the PostgreSQL test backend.
+func isPostgresTestStore(any) bool {
+	return false
+}
+
 // NewTestStore creates a new SQLite database for testing with migrations
 // applied. Each test gets its own temporary database file.
 func NewTestStore(t *testing.T) *sqlite.Store {

wallet/internal/db/itest/tx_corruption_pg_test.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/chainhash/v2"
-	"github.com/btcsuite/btcwallet/wallet/internal/db/pg"
 	"github.com/stretchr/testify/require"
 )
 
@@ -19,7 +18,7 @@ import (
 // corruptTransactionStatus writes an invalid tx status after dropping the
 // validating constraints that normally reject it. The corruption itests use
 // this to verify that reads reject impossible tx states.
-func corruptTransactionStatus(t *testing.T, store *pg.Store,
+func corruptTransactionStatus(t *testing.T, store *postgresTestStore,
 	walletID uint32, txHash chainhash.Hash, status int64) {
 
 	t.Helper()
@@ -55,7 +54,7 @@ func corruptTransactionStatus(t *testing.T, store *pg.Store,
 // corruptTransactionHash writes malformed tx-hash bytes after dropping the
 // fixed-length hash check. The corruption itests then verify that hash
 // decoding fails with the expected error path.
-func corruptTransactionHash(t *testing.T, store *pg.Store,
+func corruptTransactionHash(t *testing.T, store *postgresTestStore,
 	walletID uint32, txHash chainhash.Hash, hash []byte) {
 
 	t.Helper()
@@ -82,7 +81,7 @@ func corruptTransactionHash(t *testing.T, store *pg.Store,
 
 // forceRollbackBlockDeleteFailure installs a trigger that fails the rollback
 // block-deletion stage after sync-state rewind has run.
-func forceRollbackBlockDeleteFailure(t *testing.T, store *pg.Store) {
+func forceRollbackBlockDeleteFailure(t *testing.T, store *postgresTestStore) {
 	t.Helper()
 
 	statements := []string{
@@ -104,7 +103,7 @@ func forceRollbackBlockDeleteFailure(t *testing.T, store *pg.Store) {
 // the non-negative height check and creating a matching block row. The
 // corruption itests use this to verify that reads reject impossible
 // confirmation metadata.
-func corruptTransactionBlockHeight(t *testing.T, store *pg.Store,
+func corruptTransactionBlockHeight(t *testing.T, store *postgresTestStore,
 	walletID uint32, txHash chainhash.Hash, height int64) {
 
 	t.Helper()
@@ -144,7 +143,7 @@ func corruptTransactionBlockHeight(t *testing.T, store *pg.Store,
 // corruptUtxoOutputIndex writes an invalid output index after dropping the
 // non-negative output-index check. The corruption itests then verify that UTXO
 // decoding rejects the malformed persisted value.
-func corruptUtxoOutputIndex(t *testing.T, store *pg.Store,
+func corruptUtxoOutputIndex(t *testing.T, store *postgresTestStore,
 	walletID uint32, txHash chainhash.Hash, oldIndex uint32, newIndex int64) {
 
 	t.Helper()
@@ -172,7 +171,7 @@ func corruptUtxoOutputIndex(t *testing.T, store *pg.Store,
 // corruptActiveLeaseLockID writes an invalid lease lock ID after dropping the
 // fixed-length lock-id check. The corruption itests use this to verify that
 // lease reads reject malformed lock identifiers.
-func corruptActiveLeaseLockID(t *testing.T, store *pg.Store,
+func corruptActiveLeaseLockID(t *testing.T, store *postgresTestStore,
 	walletID uint32, txHash chainhash.Hash, outputIndex uint32, lockID []byte) {
 
 	t.Helper()

wallet/internal/db/itest/txstore_corruption_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/btcsuite/btcd/address/v2"
 	"github.com/btcsuite/btcd/wire/v2"
 	"github.com/btcsuite/btcwallet/wallet/internal/db"
-	"github.com/btcsuite/btcwallet/wallet/internal/db/pg"
 	"github.com/stretchr/testify/require"
 )
 
@@ -22,7 +21,7 @@ func dropTableForCorruption(t *testing.T, store interface{ DB() *sql.DB },
 	t.Helper()
 
 	stmt := "DROP TABLE " + table
-	if _, ok := any(store).(*pg.Store); ok {
+	if isPostgresTestStore(store) {
 		stmt += " CASCADE"
 	}