Update pgx (#3)

Makefile

@@ -0,0 +1,7 @@
+start-postgres:
+	podman stop dbump-postgres
+	podman rm dbump-postgres
+	podman run --name dbump-postgres -p 5432:5432 -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=postgres -d postgres
+
+test-pgx:
+	cd dbump_pgx && go test -v -race -shuffle=on ./...

dbump_pgx/go.mod

@@ -3,6 +3,6 @@ module github.com/cristalhq/dbump/dbump_pgx
 go 1.16
 
 require (
-	github.com/cristalhq/dbump v0.2.0
+	github.com/cristalhq/dbump v0.7.1
 	github.com/jackc/pgx/v4 v4.16.1
 )

dbump_pgx/go.sum

@@ -5,8 +5,8 @@ github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMe
 github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
-github.com/cristalhq/dbump v0.2.0 h1:zwk0d4UFBGtXswYh6lAZ5KDCEgnn9px9HMzLCUQSDso=
-github.com/cristalhq/dbump v0.2.0/go.mod h1:rAjULuStbuNPCLrJT62Eu7Sp/2gVt/4URUvsnPK1yFA=
+github.com/cristalhq/dbump v0.7.1 h1:zKXPyU1YR+2FFIs5EMo4pBWYVsI0GD+cV+JNJ6kyl14=
+github.com/cristalhq/dbump v0.7.1/go.mod h1:rAjULuStbuNPCLrJT62Eu7Sp/2gVt/4URUvsnPK1yFA=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -127,6 +127,8 @@ golang.org/x/crypto v0.0.0-20201203163018-be400aefbc4c/go.mod h1:jdWPYTVW3xRLrWP
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97 h1:/UOmuWzQfxxo9UtlXMwuQU8CMgg1eZXqTRwkSQJWKOI=
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d h1:sK3txAijHtOK88l68nt020reeT1ZdKLIYetKl95FzVY=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
@@ -135,6 +137,7 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -147,6 +150,7 @@ golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

dbump_pgx/pgx.go

@@ -2,6 +2,8 @@ package dbump_pgx
 
 import (
 	"context"
+	"errors"
+	"fmt"
 
 	"github.com/cristalhq/dbump"
 	"github.com/jackc/pgx/v4"
@@ -14,58 +16,107 @@ var _ dbump.Migrator = &Migrator{}
 
 // Migrator to migrate Postgres.
 type Migrator struct {
-	conn *pgx.Conn
+	conn      *pgx.Conn
+	tx        pgx.Tx
+	tableName string
+}
+
+// Config for the migrator.
+type Config struct {
+	// Schema for the dbump version table. Default is empty which means "public" schema.
+	Schema string
+	// Table for the dbump version table. Default is empty which means "_dbump_log" table.
+	Table string
 }
 
 // NewMigrator instantiates new Migrator.
-func NewMigrator(conn *pgx.Conn) *Migrator {
+func NewMigrator(conn *pgx.Conn, cfg Config) *Migrator {
+	if cfg.Schema != "" {
+		cfg.Schema += "."
+	}
+	if cfg.Table == "" {
+		cfg.Table = "_dbump_log"
+	}
 	return &Migrator{
-		conn: conn,
+		conn:      conn,
+		tableName: cfg.Schema + cfg.Table,
 	}
 }
 
-// Init migrator.
+// Init is a method from Migrator interface.
 func (pg *Migrator) Init(ctx context.Context) error {
-	query := `CREATE TABLE IF NOT EXISTS _dbump_schema_version (
-	version    BIGINT                   NOT NULL PRIMARY KEY,
+	query := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
+	version    BIGINT NOT NULL,
 	created_at TIMESTAMP WITH TIME ZONE NOT NULL
-);`
+);`, pg.tableName)
 	_, err := pg.conn.Exec(ctx, query)
 	return err
 }
 
-// LockDB is a method for Migrator interface.
+// LockDB is a method from Migrator interface.
 func (pg *Migrator) LockDB(ctx context.Context) error {
 	_, err := pg.conn.Exec(ctx, "SELECT pg_advisory_lock($1);", lockNum)
 	return err
 }
 
-// UnlockDB is a method for Migrator interface.
+// UnlockDB is a method from Migrator interface.
 func (pg *Migrator) UnlockDB(ctx context.Context) error {
 	_, err := pg.conn.Exec(ctx, "SELECT pg_advisory_unlock($1);", lockNum)
 	return err
 }
 
-// Version is a method for Migrator interface.
+// Version is a method from Migrator interface.
 func (pg *Migrator) Version(ctx context.Context) (version int, err error) {
-	query := "SELECT COUNT(*) FROM _dbump_schema_version;"
-	row := pg.conn.QueryRow(ctx, query)
+	query := fmt.Sprintf("SELECT version FROM %s ORDER BY created_at DESC LIMIT 1;", pg.tableName)
+	var row pgx.Row
+	if pg.tx != nil {
+		row = pg.tx.QueryRow(ctx, query)
+	} else {
+		row = pg.conn.QueryRow(ctx, query)
+	}
 	err = row.Scan(&version)
+	if err != nil && errors.Is(err, pgx.ErrNoRows) {
+		return 0, nil
+	}
 	return version, err
 }
 
-// SetVersion is a method for Migrator interface.
+// SetVersion is a method from Migrator interface.
 func (pg *Migrator) SetVersion(ctx context.Context, version int) error {
-	query := `INSERT INTO _dbump_schema_version (version, created_at)
-VALUES ($1, NOW())
-ON CONFLICT (version) DO UPDATE
-SET created_at = NOW();`
-	_, err := pg.conn.Exec(ctx, query, version)
+	query := fmt.Sprintf("INSERT INTO %s (version, created_at) VALUES ($1, NOW());", pg.tableName)
+	var err error
+	if pg.tx != nil {
+		_, err = pg.tx.Exec(ctx, query, version)
+	} else {
+		_, err = pg.conn.Exec(ctx, query, version)
+	}
 	return err
 }
 
-// Exec is a method for Migrator interface.
+// Begin is a method from Migrator interface.
+func (pg *Migrator) Begin(ctx context.Context) error {
+	var err error
+	pg.tx, err = pg.conn.Begin(ctx)
+	return err
+}
+
+// Commit is a method from Migrator interface.
+func (pg *Migrator) Commit(ctx context.Context) error {
+	return pg.tx.Commit(ctx)
+}
+
+// Rollback is a method from Migrator interface.
+func (pg *Migrator) Rollback(ctx context.Context) error {
+	return pg.tx.Rollback(ctx)
+}
+
+// Exec is a method from Migrator interface.
 func (pg *Migrator) Exec(ctx context.Context, query string, args ...interface{}) error {
-	_, err := pg.conn.Exec(ctx, query, args...)
+	var err error
+	if pg.tx != nil {
+		_, err = pg.tx.Exec(ctx, query, args...)
+	} else {
+		_, err = pg.conn.Exec(ctx, query, args...)
+	}
 	return err
 }