Upgrade pgx to v4 (heroiclabs#615)

Export db stats to prometheus

go.mod

@@ -5,7 +5,6 @@ go 1.16
 require (
 	github.com/blevesearch/bleve/v2 v2.0.3
 	github.com/blevesearch/upsidedown_store_api v1.0.1
-	github.com/cockroachdb/apd v1.1.0 // indirect
 	github.com/dgrijalva/jwt-go v3.2.1-0.20200107013213-dc14462fd587+incompatible
 	github.com/dop251/goja v0.0.0-20210406175830-1b11a6af686d
 	github.com/glycerine/go-unsnap-stream v0.0.0-20190901134440-81cf024a9e0a // indirect
@@ -15,23 +14,23 @@ require (
 	github.com/gorilla/websocket v1.4.2
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.3.0
 	github.com/heroiclabs/nakama-common v1.14.0
-	github.com/jackc/fake v0.0.0-20150926172116-812a484cc733 // indirect
-	github.com/jackc/pgx v3.5.0+incompatible
+	github.com/jackc/pgconn v1.8.1
+	github.com/jackc/pgerrcode v0.0.0-20201024163028-a0d42d470451
+	github.com/jackc/pgtype v1.7.0
+	github.com/jackc/pgx/v4 v4.11.0
 	github.com/m3db/prometheus_client_golang v0.8.1 // indirect
 	github.com/m3db/prometheus_client_model v0.1.0 // indirect
 	github.com/m3db/prometheus_common v0.1.0 // indirect
 	github.com/m3db/prometheus_procfs v0.8.1 // indirect
 	github.com/rubenv/sql-migrate v0.0.0-20210408115534-a32ed26c37ea
-	github.com/satori/go.uuid v1.2.0 // indirect
-	github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24 // indirect
 	github.com/steveyen/gtreap v0.1.0
 	github.com/stretchr/testify v1.7.0
 	github.com/tinylib/msgp v1.1.2 // indirect
 	github.com/uber-go/tally v3.3.17+incompatible
 	github.com/ziutek/mymysql v1.5.4 // indirect
 	go.uber.org/atomic v1.7.0
 	go.uber.org/zap v1.16.0
-	golang.org/x/crypto v0.0.0-20210415154028-4f45737414dc
+	golang.org/x/crypto v0.0.0-20210506145944-38f3c27a63bf
 	google.golang.org/genproto v0.0.0-20210224155714-063164c882e6
 	google.golang.org/grpc v1.37.0
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0

main.go

@@ -34,12 +34,11 @@ import (
 	"path/filepath"
 
 	"github.com/gofrs/uuid"
-
 	"github.com/heroiclabs/nakama/v3/ga"
 	"github.com/heroiclabs/nakama/v3/migrate"
 	"github.com/heroiclabs/nakama/v3/server"
 	"github.com/heroiclabs/nakama/v3/social"
-	_ "github.com/jackc/pgx/stdlib"
+	_ "github.com/jackc/pgx/v4/stdlib"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
@@ -108,12 +107,12 @@ func main() {
 
 	redactedAddresses := make([]string, 0, 1)
 	for _, address := range config.GetDatabase().Addresses {
-		rawURL := fmt.Sprintf("postgresql://%s", address)
+		rawURL := fmt.Sprintf("postgres://%s", address)
 		parsedURL, err := url.Parse(rawURL)
 		if err != nil {
 			logger.Fatal("Bad connection URL", zap.Error(err))
 		}
-		redactedAddresses = append(redactedAddresses, strings.TrimPrefix(parsedURL.Redacted(), "postgresql://"))
+		redactedAddresses = append(redactedAddresses, strings.TrimPrefix(parsedURL.Redacted(), "postgres://"))
 	}
 	startupLogger.Info("Database connections", zap.Strings("dsns", redactedAddresses))
 
@@ -128,7 +127,7 @@ func main() {
 
 	// Start up server components.
 	cookie := newOrLoadCookie(config)
-	metrics := server.NewMetrics(logger, startupLogger, config)
+	metrics := server.NewMetrics(logger, startupLogger, db, config)
 	sessionRegistry := server.NewLocalSessionRegistry(metrics)
 	sessionCache := server.NewLocalSessionCache(config)
 	statusRegistry := server.NewStatusRegistry(logger, config, sessionRegistry, jsonpbMarshaler)

migrate/migrate.go

@@ -17,8 +17,12 @@ package migrate
 import (
 	"database/sql"
 	"embed"
+	"errors"
 	"flag"
 	"fmt"
+	"github.com/jackc/pgconn"
+	"github.com/jackc/pgerrcode"
+	_ "github.com/jackc/pgx/v4/stdlib"
 	"io/ioutil"
 	"math"
 	"net/url"
@@ -27,8 +31,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/jackc/pgx"
-	_ "github.com/jackc/pgx/stdlib" // Blank import to register SQL driver
 	migrate "github.com/rubenv/sql-migrate"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -37,8 +39,7 @@ import (
 )
 
 const (
-	dbErrorDatabaseDoesNotExist = "3D000"
-	dbErrorDuplicateDatabase    = "42P04"
+	dbErrorDatabaseDoesNotExist = pgerrcode.InvalidCatalogName
 	migrationTable              = "migration_info"
 	dialect                     = "postgres"
 	defaultLimit                = -1
@@ -185,39 +186,47 @@ func Parse(args []string, tmpLogger *zap.Logger) {
 		logger.Fatal("Failed to open database", zap.Error(err))
 	}
 
-	var dbVersion string
-	if err = db.QueryRow("SELECT version()").Scan(&dbVersion); err != nil {
-		if e, ok := err.(pgx.PgError); ok && e.Code == dbErrorDatabaseDoesNotExist {
-			// Database does not exist, try to create a new one
-			logger.Info("Creating new database", zap.String("name", dbname))
+	var nakamaDBExists bool
+	if err = db.QueryRow("SELECT EXISTS (SELECT 1 from pg_database WHERE datname = $1)", dbname).Scan(&nakamaDBExists); err != nil {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) && pgErr.Code == dbErrorDatabaseDoesNotExist {
+			nakamaDBExists = false
+		} else {
 			db.Close()
-			// Connect to anonymous db
-			parsedURL.Path = ""
-			db, err = sql.Open("pgx", parsedURL.String())
-			if err != nil {
-				logger.Fatal("Failed to open database", zap.Error(err))
-			}
-			if _, err = db.Exec(fmt.Sprintf("CREATE DATABASE %s", dbname)); err != nil {
-				db.Close()
-				logger.Fatal("Failed to create database", zap.Error(err))
-			}
+			logger.Fatal("Failed to check if db exists", zap.String("db", dbname), zap.Error(err))
+		}
+	}
+
+	if !nakamaDBExists {
+		// Database does not exist, create it
+		logger.Info("Creating new database", zap.String("name", dbname))
+		db.Close()
+		// Connect to anonymous db
+		parsedURL.Path = ""
+		db, err = sql.Open("pgx", parsedURL.String())
+		if err != nil {
+			logger.Fatal("Failed to open database", zap.Error(err))
+		}
+		if _, err = db.Exec(fmt.Sprintf("CREATE DATABASE %s", dbname)); err != nil {
 			db.Close()
-			parsedURL.Path = fmt.Sprintf("/%s", dbname)
-			db, err = sql.Open("pgx", parsedURL.String())
-			if err != nil {
-				db.Close()
-				logger.Fatal("Failed to open database", zap.Error(err))
-			}
-			// Reattempt to get database version
-			if err = db.QueryRow("SELECT version()").Scan(&dbVersion); err != nil {
-				db.Close()
-				logger.Fatal("Error querying database version", zap.Error(err))
-			}
-		} else {
+			logger.Fatal("Failed to create database", zap.Error(err))
+		}
+		db.Close()
+		parsedURL.Path = fmt.Sprintf("/%s", dbname)
+		db, err = sql.Open("pgx", parsedURL.String())
+		if err != nil {
 			db.Close()
-			logger.Fatal("Error querying database version", zap.Error(err))
+			logger.Fatal("Failed to open database", zap.Error(err))
 		}
 	}
+
+	// Get database version
+	var dbVersion string
+	if err = db.QueryRow("SELECT version()").Scan(&dbVersion); err != nil {
+		db.Close()
+		logger.Fatal("Error querying database version", zap.Error(err))
+	}
+
 	logger.Info("Database information", zap.String("version", dbVersion))
 
 	if err = db.Ping(); err != nil {

server/api_account.go

@@ -16,9 +16,10 @@ package server
 
 import (
 	"context"
+	"errors"
 	"github.com/gofrs/uuid"
 	"github.com/heroiclabs/nakama-common/api"
-	"github.com/jackc/pgx"
+	"github.com/jackc/pgconn"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -114,7 +115,8 @@ func (s *ApiServer) UpdateAccount(ctx context.Context, in *api.UpdateAccountRequ
 		metadata:    nil,
 	}})
 	if err != nil {
-		if _, ok := err.(pgx.PgError); ok {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) {
 			return nil, status.Error(codes.Internal, "Error while trying to update account.")
 		}
 		return nil, status.Error(codes.InvalidArgument, err.Error())

server/api_test.go

@@ -28,7 +28,7 @@ import (
 	"github.com/heroiclabs/nakama-common/rtapi"
 	"github.com/heroiclabs/nakama-common/runtime"
 	"github.com/heroiclabs/nakama/v3/apigrpc"
-	_ "github.com/jackc/pgx/stdlib"
+	_ "github.com/jackc/pgx/v4/stdlib"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"google.golang.org/grpc"

server/console_account.go

@@ -25,7 +25,7 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/jackc/pgx/pgtype"
+	"github.com/jackc/pgtype"
 	"golang.org/x/crypto/bcrypt"
 
 	"github.com/gofrs/uuid"

server/console_authenticate.go

@@ -23,7 +23,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/jackc/pgx/pgtype"
+	"github.com/jackc/pgtype"
 	"go.uber.org/zap"
 	"golang.org/x/crypto/bcrypt"
 	"google.golang.org/grpc/codes"

server/console_storage.go

@@ -28,7 +28,7 @@ import (
 	"github.com/gofrs/uuid"
 	"github.com/heroiclabs/nakama-common/api"
 	"github.com/heroiclabs/nakama/v3/console"
-	"github.com/jackc/pgx/pgtype"
+	"github.com/jackc/pgtype"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"

server/console_user.go

@@ -18,13 +18,14 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
+	"github.com/jackc/pgconn"
 	"net/http"
 	"regexp"
 	"unicode"
 
 	"github.com/gofrs/uuid"
 	"github.com/heroiclabs/nakama/v3/console"
-	"github.com/jackc/pgx"
 	"go.uber.org/zap"
 	"golang.org/x/crypto/bcrypt"
 	"google.golang.org/grpc/codes"
@@ -101,8 +102,9 @@ func (s *ConsoleServer) dbInsertConsoleUser(ctx context.Context, in *console.Add
 	query := "INSERT INTO console_user (id, username, email, password, role) VALUES ($1, $2, $3, $4, $5)"
 	_, err = s.db.ExecContext(ctx, query, id.String(), in.Username, in.Email, hashedPassword, in.Role)
 	if err != nil {
-		if perr, is := err.(pgx.PgError); is {
-			if perr.Code == dbErrorUniqueViolation {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) {
+			if pgErr.Code == dbErrorUniqueViolation {
 				return false, nil
 			}
 		}

server/core_account.go

@@ -22,8 +22,8 @@ import (
 	"github.com/gofrs/uuid"
 	"github.com/heroiclabs/nakama-common/api"
 	"github.com/heroiclabs/nakama/v3/console"
-	"github.com/jackc/pgx"
-	"github.com/jackc/pgx/pgtype"
+	"github.com/jackc/pgconn"
+	"github.com/jackc/pgtype"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -351,7 +351,8 @@ func updateAccounts(ctx context.Context, logger *zap.Logger, tx *sql.Tx, updates
 			" WHERE id = $1 AND (" + strings.Join(distinctStatements, " OR ") + ")"
 
 		if _, err := tx.ExecContext(ctx, query, params...); err != nil {
-			if e, ok := err.(pgx.PgError); ok && e.Code == dbErrorUniqueViolation && strings.Contains(e.Message, "users_username_key") {
+			var pgErr *pgconn.PgError
+			if errors.As(err, &pgErr) && pgErr.Code == dbErrorUniqueViolation && strings.Contains(pgErr.Message, "users_username_key") {
 				return errors.New("Username is already in use.")
 			}