.github/dependabot.yml

@@ -1,10 +1,18 @@
 version: 2
 updates:
-  - package-ecosystem: gomod
+  - package-ecosystem: "gomod"
+    commit-message:
+      prefix: "deps:"
     directory: "/"
     schedule:
-      interval: daily
-  - package-ecosystem: github-actions
+      interval: "weekly"
+      day: "sunday"
+      time: "09:00"
+  - package-ecosystem: "github-actions"
+    commit-message:
+      prefix: "ci:"
     directory: "/"
     schedule:
-      interval: daily
+      interval: "weekly"
+      day: "sunday"
+      time: "09:00"

.github/workflows/build.yml

@@ -4,20 +4,14 @@ on:
   push:
     branches: [main]
   pull_request:
-  workflow_dispatch:
-    inputs:
-      tag:
-        description: 'Tag to create'
-        required: true
-        default: 'v0.0.0'
+    branches: [main]
+  schedule:
+    - cron: '0 0 * * 0' # run "At 00:00 on Sunday"
 
 # See https://github.com/cristalhq/.github/.github/workflows
 jobs:
   build:
-    uses: cristalhq/.github/.github/workflows/build.yml@main
+    uses: cristalhq/.github/.github/workflows/build.yml@v0.8.1
 
-  release:
-    if: github.event_name == 'workflow_dispatch'
-    uses: cristalhq/.github/.github/workflows/release.yml@main
-    with:
-      tag: ${{ github.event.input.tag }}
+  vuln:
+    uses: cristalhq/.github/.github/workflows/vuln.yml@v0.8.1

README.md

@@ -2,8 +2,6 @@
 
 [![build-img]][build-url]
 [![pkg-img]][pkg-url]
-[![reportcard-img]][reportcard-url]
-[![coverage-img]][coverage-url]
 [![version-img]][version-url]
 
 Go database schema migrator library (See [cristalhq/dbumper](https://github.com/cristalhq/dbumper) tool).
@@ -29,7 +27,7 @@ See [GUIDE.md](https://github.com/cristalhq/dbump/blob/main/GUIDE.md) for more d
 
 ## Install
 
-Go version 1.16+
+Go version 1.17+
 
 ```
 go get github.com/cristalhq/dbump
@@ -66,9 +64,5 @@ See [these docs][pkg-url].
 [build-url]: https://github.com/cristalhq/dbump/actions
 [pkg-img]: https://pkg.go.dev/badge/cristalhq/dbump
 [pkg-url]: https://pkg.go.dev/github.com/cristalhq/dbump
-[reportcard-img]: https://goreportcard.com/badge/cristalhq/dbump
-[reportcard-url]: https://goreportcard.com/report/cristalhq/dbump
-[coverage-img]: https://codecov.io/gh/cristalhq/dbump/branch/main/graph/badge.svg
-[coverage-url]: https://codecov.io/gh/cristalhq/dbump
 [version-img]: https://img.shields.io/github/v/release/cristalhq/dbump
 [version-url]: https://github.com/cristalhq/dbump/releases

dbump.go

@@ -14,10 +14,10 @@ import (
 var ErrMigrationAlreadyLocked = errors.New("migration is locked already")
 
 // MigrationDelimiter separates apply and revert queries inside a migration step/file.
-// Is exported just to be used by https://github.com/cristalhq/dbumper
+// Const is exported to be used by https://github.com/cristalhq/dbumper tool.
 const MigrationDelimiter = `--- apply above / revert below ---`
 
-// Config for the migration process. Is used only by Run function.
+// Config of the migration process. Used by Run function.
 type Config struct {
 	// Migrator represents a database.
 	Migrator Migrator
@@ -44,7 +44,7 @@ type Config struct {
 
 	// DisableTx will run every migration not in a transaction.
 	// This completely depends on a specific Migrator implementation
-	// because not every database supports transaction, so this option can be no-op all the time.
+	// because not every database supports transaction, so this option can be no-op for some databases.
 	DisableTx bool
 
 	// UseForce to get a lock on a database. MUST be used with the caution.
@@ -60,9 +60,12 @@ type Config struct {
 	// BeforeStep function will be invoked right before the DoStep for each step.
 	// Default is nil and means no-op.
 	BeforeStep func(ctx context.Context, step Step)
+
 	// AfterStep function will be invoked right after the DoStep for each step.
 	// Default is nil and means no-op.
 	AfterStep func(ctx context.Context, step Step)
+
+	_ struct{} // enforce explicit field names.
 }
 
 // Migrator represents database over which we will run migrations.
@@ -72,7 +75,7 @@ type Migrator interface {
 	// UnlockDB to allow running other migrators later.
 	UnlockDB(ctx context.Context) error
 
-	// Init the dbump database where database state is saved.
+	// Init the dbump table where database state is saved.
 	// What is created by this method completely depends on migrator implementation
 	// and might be different between databases.
 	Init(ctx context.Context) error
@@ -103,9 +106,9 @@ type Loader interface {
 // Migration represents migration step that will be runned on a database.
 type Migration struct {
 	ID     int    // ID of the migration, unique, positive, starts from 1.
-	Name   string // Name of the migration
-	Apply  string // Apply query
-	Revert string // Revert query
+	Name   string // Name of the migration.
+	Apply  string // Apply query.
+	Revert string // Revert query.
 }
 
 // MigratorMode to change migration flow.
@@ -131,17 +134,17 @@ func Run(ctx context.Context, config Config) error {
 		return errors.New("loader cannot be nil")
 	case config.Mode == ModeNotSet:
 		return errors.New("mode not set")
-	case config.Mode >= modeMaxPossible:
+	case config.Mode < 0 || config.Mode >= modeMaxPossible:
 		return fmt.Errorf("incorrect mode provided: %d", config.Mode)
 	case config.Num <= 0 && (config.Mode == ModeApplyN || config.Mode == ModeRevertN):
 		return fmt.Errorf("num must be greater than 0: %d", config.Num)
 	}
 
 	if config.BeforeStep == nil {
-		config.BeforeStep = func(ctx context.Context, step Step) {}
+		config.BeforeStep = noopHook
 	}
 	if config.AfterStep == nil {
-		config.AfterStep = func(ctx context.Context, step Step) {}
+		config.AfterStep = noopHook
 	}
 
 	m := mig{
@@ -195,7 +198,8 @@ func (m *mig) runMigrations(ctx context.Context, ms []*Migration) (err error) {
 	}
 
 	defer func() {
-		if errUnlock := m.unlockDB(ctx); err == nil && errUnlock != nil {
+		errUnlock := m.unlockDB(ctx)
+		if err == nil && errUnlock != nil {
 			err = fmt.Errorf("unlock db: %w", errUnlock)
 		}
 	}()
@@ -206,6 +210,7 @@ func (m *mig) runMigrations(ctx context.Context, ms []*Migration) (err error) {
 	}
 	err = m.runMigrationsLocked(ctx, ms)
 
+	// drop all dbump data.
 	if m.Mode == ModeDrop {
 		err = m.Drop(ctx)
 	}
@@ -244,11 +249,12 @@ func (m *mig) runMigrationsLocked(ctx context.Context, ms []*Migration) error {
 		return fmt.Errorf("version get: %w", err)
 	}
 
-	for i, step := range m.prepareSteps(curr, target, ms) {
+	steps := m.prepareSteps(curr, target, ms)
+	for i, step := range steps {
 		m.BeforeStep(ctx, step)
 
 		if err := m.step(ctx, step); err != nil {
-			return fmt.Errorf("migration %d: %w", i, err)
+			return fmt.Errorf("migration %d: %w\n%s", i, err, step.Query)
 		}
 
 		m.AfterStep(ctx, step)
@@ -265,6 +271,7 @@ func (m *mig) step(ctx context.Context, step Step) error {
 	return m.DoStep(ctx, step)
 }
 
+// getCurrAndTargetVersions returns current version of the schema and the target version based on run config.
 func (m *mig) getCurrAndTargetVersions(ctx context.Context, migrations int) (curr, target int, err error) {
 	curr, err = m.Version(ctx)
 	if err != nil {
@@ -313,6 +320,7 @@ func (m *mig) getCurrAndTargetVersions(ctx context.Context, migrations int) (cur
 
 func (m *mig) prepareSteps(curr, target int, ms []*Migration) []Step {
 	if m.Mode == ModeRedo {
+		// undo & do current step.
 		return []Step{
 			ms[curr-1].toStep(false, m.DisableTx),
 			ms[curr-1].toStep(true, m.DisableTx),
@@ -360,3 +368,5 @@ func (m *Migration) toStep(up, disableTx bool) Step {
 		DisableTx: disableTx,
 	}
 }
+
+func noopHook(context.Context, Step) {}

dbump_pg/go.mod

@@ -3,7 +3,7 @@ module github.com/cristalhq/dbump/dbump_pg
 go 1.17
 
 require (
-	github.com/cristalhq/dbump v0.3.0
+	github.com/cristalhq/dbump v0.14.0
 	github.com/lib/pq v1.10.6
 )
 

dbump_pg/pg.go

@@ -3,70 +3,136 @@ package dbump_pg
 import (
 	"context"
 	"database/sql"
+	"errors"
+	"fmt"
+	"hash/fnv"
 
 	"github.com/cristalhq/dbump"
 )
 
-// to prevent multiple migrations running at the same time
-const lockNum int64 = 707_707_707
-
 var _ dbump.Migrator = &Migrator{}
 
 // Migrator to migrate Postgres.
 type Migrator struct {
 	conn *sql.DB
+	cfg  Config
+}
+
+// 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
+
+	// [schema.]table
+	tableName string
+	// to prevent multiple migrations running at the same time
+	lockNum int64
 }
 
 // NewMigrator instantiates new Migrator.
 // Takes std *sql.DB.
-func NewMigrator(conn *sql.DB) *Migrator {
+func NewMigrator(conn *sql.DB, cfg Config) *Migrator {
+	if cfg.Schema == "" {
+		cfg.Schema = "public"
+	}
+	if cfg.Table == "" {
+		cfg.Table = "_dbump_log"
+	}
+
+	cfg.tableName = cfg.Schema + "." + cfg.Table
+	cfg.lockNum = hashTableName(cfg.tableName)
+
 	return &Migrator{
 		conn: conn,
+		cfg:  cfg,
 	}
 }
 
 // Init migrator.
 func (pg *Migrator) Init(ctx context.Context) error {
-	query := `CREATE TABLE IF NOT EXISTS _dbump_schema_version (
-	version    BIGINT                   NOT NULL PRIMARY KEY,
+	var query string
+	if pg.cfg.Schema != "" {
+		query = fmt.Sprintf(`CREATE SCHEMA IF NOT EXISTS %s;`, pg.cfg.Schema)
+	}
+
+	query += fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
+	version    BIGINT NOT NULL,
 	created_at TIMESTAMP WITH TIME ZONE NOT NULL
-);`
+);`, pg.cfg.tableName)
+
+	_, err := pg.conn.ExecContext(ctx, query)
+	return err
+}
+
+// Drop is a method from Migrator interface.
+func (pg *Migrator) Drop(ctx context.Context) error {
+	query := fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, pg.cfg.tableName)
 	_, err := pg.conn.ExecContext(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.ExecContext(ctx, "SELECT pg_advisory_lock($1);", lockNum)
+	_, err := pg.conn.ExecContext(ctx, "SELECT pg_advisory_lock($1);", pg.cfg.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.ExecContext(ctx, "SELECT pg_advisory_unlock($1);", lockNum)
+	_, err := pg.conn.ExecContext(ctx, "SELECT pg_advisory_unlock($1);", pg.cfg.lockNum)
 	return err
 }
 
 // Version is a method for Migrator interface.
 func (pg *Migrator) Version(ctx context.Context) (version int, err error) {
-	query := "SELECT COUNT(*) FROM _dbump_schema_version;"
+	query := fmt.Sprintf("SELECT version FROM %s ORDER BY created_at DESC LIMIT 1;", pg.cfg.tableName)
 	row := pg.conn.QueryRowContext(ctx, query)
 	err = row.Scan(&version)
+	if err != nil && errors.Is(err, sql.ErrNoRows) {
+		return 0, nil
+	}
 	return version, err
 }
 
-// SetVersion is a method for 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.ExecContext(ctx, query, version)
-	return err
+// DoStep is a method for Migrator interface.
+func (pg *Migrator) DoStep(ctx context.Context, step dbump.Step) error {
+	if step.DisableTx {
+		if _, err := pg.conn.ExecContext(ctx, step.Query); err != nil {
+			return err
+		}
+		query := fmt.Sprintf("INSERT INTO %s (version, created_at) VALUES ($1, NOW());", pg.cfg.tableName)
+		_, err := pg.conn.ExecContext(ctx, query, step.Version)
+		return err
+	}
+
+	return pg.beginFunc(ctx, func(tx *sql.Tx) error {
+		if _, err := tx.ExecContext(ctx, step.Query); err != nil {
+			return err
+		}
+		query := fmt.Sprintf("INSERT INTO %s (version, created_at) VALUES ($1, NOW());", pg.cfg.tableName)
+		_, err := tx.ExecContext(ctx, query, step.Version)
+		return err
+	})
 }
 
-// Exec is a method for Migrator interface.
-func (pg *Migrator) Exec(ctx context.Context, query string, args ...interface{}) error {
-	_, err := pg.conn.ExecContext(ctx, query, args...)
-	return err
+func (pg *Migrator) beginFunc(ctx context.Context, f func(*sql.Tx) error) (err error) {
+	tx, err := pg.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	if err := f(tx); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
+func hashTableName(s string) int64 {
+	h := fnv.New64()
+	h.Write([]byte(s))
+	return int64(h.Sum64())
 }