Initial implementation (extracted from dbumper)

Author: cristaloleg

.github/workflows/build.yml

@@ -0,0 +1,55 @@
+name: build
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+  workflow_dispatch:
+    inputs:
+      tag:
+        description: 'Tag to create'
+        required: true
+        default: 'v0.0.0'
+
+jobs:
+
+  build:
+    name: Build & Test
+    runs-on: ubuntu-latest
+    steps:
+    - name: Set up Go 1.x
+      uses: actions/setup-go@v2
+      with:
+        go-version: ^1.x
+
+    - name: Check out code
+      uses: actions/checkout@v2
+
+    - name: Test
+      run: go test -v -coverprofile=coverage.txt ./...
+
+    - name: Upload Coverage
+      uses: codecov/codecov-action@v1
+      continue-on-error: true
+      with:
+        token: ${{secrets.CODECOV_TOKEN}}
+        file: ./coverage.txt
+        fail_ci_if_error: false
+
+  release:
+    if: github.event_name == 'workflow_dispatch'
+
+    name: Release
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Checkout with tags
+        run: git fetch --prune --unshallow --tags
+
+      - name: Create release
+        run: |
+          git log --format="%C(auto) %H %s" `git tag --sort=-committerdate | head -1`...HEAD > changelog.txt
+          echo ${{ secrets.GITHUB_TOKEN }} | gh auth login --with-token
+          gh release create ${{ github.event.inputs.tag }} -t ${{ github.event.inputs.tag }} -F changelog.txt

README.md

@@ -1,2 +1,47 @@
 # dbump
-Go database schema migrator library (See dbumper tool)
+
+# dbumper
+
+[![build-img]][build-url]
+[![pkg-img]][pkg-url]
+[![reportcard-img]][reportcard-url]
+[![coverage-img]][coverage-url]
+
+Go database schema migrator library (See [dbumper](https://github.com/cristalhq/dbumper) tool).
+
+## Rationale
+
+TODO
+
+## Features
+
+* Simple.
+
+## Install
+
+Go version 1.16+
+
+```
+go get github.com/cristalhq/dbumper
+```
+
+## Example
+
+TODO
+
+## Documentation
+
+See [these docs][pkg-url].
+
+## License
+
+[MIT License](LICENSE).
+
+[build-img]: https://github.com/cristalhq/dbumper/workflows/build/badge.svg
+[build-url]: https://github.com/cristalhq/dbumper/actions
+[pkg-img]: https://pkg.go.dev/badge/cristalhq/dbumper
+[pkg-url]: https://pkg.go.dev/github.com/cristalhq/dbumper
+[reportcard-img]: https://goreportcard.com/badge/cristalhq/dbumper
+[reportcard-url]: https://goreportcard.com/report/cristalhq/dbumper
+[coverage-img]: https://codecov.io/gh/cristalhq/dbumper/branch/main/graph/badge.svg
+[coverage-url]: https://codecov.io/gh/cristalhq/dbumper

db_clickhouse.go

@@ -0,0 +1,51 @@
+package dbump
+
+import (
+	"context"
+	"database/sql"
+)
+
+// MigratorClickHouse to migrate ClickHouse.
+type MigratorClickHouse struct {
+	db           *sql.DB
+	versionTable string
+}
+
+// NewMigratorClickHouse instantiates new MigratorClickHouse.
+func NewMigratorClickHouse(db *sql.DB) *MigratorClickHouse {
+	return &MigratorClickHouse{
+		db:           db,
+		versionTable: "_schema_version",
+	}
+}
+
+// Lock is a method for Migrator interface.
+func (ch *MigratorClickHouse) Lock(ctx context.Context) error {
+	// TODO: currently no-op
+	return nil
+}
+
+// Unlock is a method for Migrator interface.
+func (ch *MigratorClickHouse) Unlock(ctx context.Context) error {
+	// TODO: currently no-op
+	return nil
+}
+
+// Version is a method for Migrator interface.
+func (ch *MigratorClickHouse) Version(ctx context.Context) (version int, err error) {
+	row := ch.db.QueryRowContext(ctx, "SELECT version FROM "+ch.versionTable)
+	err = row.Scan(&version)
+	return version, err
+}
+
+// SetVersion is a method for Migrator interface.
+func (ch *MigratorClickHouse) SetVersion(ctx context.Context, version int) error {
+	_, err := ch.db.ExecContext(ctx, "UPDATE "+ch.versionTable+" SET version = $1", version)
+	return err
+}
+
+// Exec is a method for Migrator interface.
+func (ch *MigratorClickHouse) Exec(ctx context.Context, query string, args ...interface{}) error {
+	_, err := ch.db.ExecContext(ctx, query)
+	return err
+}

db_mysql.go

@@ -0,0 +1,51 @@
+package dbump
+
+import (
+	"context"
+	"database/sql"
+)
+
+// MigratorMySQL to migrate MySQL.
+type MigratorMySQL struct {
+	db           *sql.DB
+	versionTable string
+}
+
+// NewMigratorMySQL instantiates new MigratorMySQL.
+func NewMigratorMySQL(db *sql.DB) *MigratorMySQL {
+	return &MigratorMySQL{
+		db:           db,
+		versionTable: "_schema_version",
+	}
+}
+
+// Lock is a method for Migrator interface.
+func (my *MigratorMySQL) Lock(ctx context.Context) error {
+	_, err := my.db.ExecContext(ctx, `SELECT GET_LOCK(?, 10)`, lockNum)
+	return err
+}
+
+// Unlock is a method for Migrator interface.
+func (my *MigratorMySQL) Unlock(ctx context.Context) error {
+	_, err := my.db.ExecContext(ctx, "SELECT RELEASE_LOCK(?)", lockNum)
+	return err
+}
+
+// Version is a method for Migrator interface.
+func (my *MigratorMySQL) Version(ctx context.Context) (version int, err error) {
+	row := my.db.QueryRowContext(ctx, "SELECT version FROM "+my.versionTable)
+	err = row.Scan(&version)
+	return version, err
+}
+
+// SetVersion is a method for Migrator interface.
+func (my *MigratorMySQL) SetVersion(ctx context.Context, version int) error {
+	_, err := my.db.ExecContext(ctx, "UPDATE "+my.versionTable+" SET version = $1", version)
+	return err
+}
+
+// Exec is a method for Migrator interface.
+func (my *MigratorMySQL) Exec(ctx context.Context, query string, args ...interface{}) error {
+	_, err := my.db.ExecContext(ctx, query)
+	return err
+}

db_postgres.go

@@ -0,0 +1,51 @@
+package dbump
+
+import (
+	"context"
+	"database/sql"
+)
+
+// MigratorPostgres to migrate Postgres.
+type MigratorPostgres struct {
+	db           *sql.DB
+	versionTable string
+}
+
+// NewMigratorPostgres instantiates new MigratorPostgres.
+func NewMigratorPostgres(db *sql.DB) *MigratorPostgres {
+	return &MigratorPostgres{
+		db:           db,
+		versionTable: "_schema_version",
+	}
+}
+
+// Lock is a method for Migrator interface.
+func (pg *MigratorPostgres) Lock(ctx context.Context) error {
+	_, err := pg.db.ExecContext(ctx, "SELECT pg_advisory_lock($1)", lockNum)
+	return err
+}
+
+// Unlock is a method for Migrator interface.
+func (pg *MigratorPostgres) Unlock(ctx context.Context) error {
+	_, err := pg.db.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", lockNum)
+	return err
+}
+
+// Version is a method for Migrator interface.
+func (pg *MigratorPostgres) Version(ctx context.Context) (version int, err error) {
+	row := pg.db.QueryRowContext(ctx, "SELECT version FROM "+pg.versionTable)
+	err = row.Scan(&version)
+	return version, err
+}
+
+// SetVersion is a method for Migrator interface.
+func (pg *MigratorPostgres) SetVersion(ctx context.Context, version int) error {
+	_, err := pg.db.ExecContext(ctx, "UPDATE "+pg.versionTable+" SET version = $1", version)
+	return err
+}
+
+// Exec is a method for Migrator interface.
+func (pg *MigratorPostgres) Exec(ctx context.Context, query string, args ...interface{}) error {
+	_, err := pg.db.ExecContext(ctx, query)
+	return err
+}

dbump.go

@@ -0,0 +1,127 @@
+package dbump
+
+import (
+	"context"
+	"fmt"
+	"sort"
+)
+
+// MigrationDelimiter separates apply and rollback queries inside a migration step/file.
+const MigrationDelimiter = `--- apply above / rollback below ---`
+
+// to prevent multiple migrations running at the same time
+const lockNum int64 = 777_777_777
+
+// Migrator represents DB over which we will run migration queries.
+type Migrator interface {
+	Lock(ctx context.Context) error
+	Unlock(ctx context.Context) error
+
+	Version(ctx context.Context) (version int, err error)
+	SetVersion(ctx context.Context, version int) error
+
+	Exec(ctx context.Context, query string, args ...interface{}) error
+}
+
+// Loader returns migrations to be applied on a DB.
+type Loader interface {
+	Load() ([]*Migration, error)
+}
+
+// Migration represents migration step that will be runned on DB.
+type Migration struct {
+	ID       int    // ID of the migration, unique, positive, starts from 1.
+	Name     string // Name of the migration
+	Apply    string // Apply query
+	Rollback string // Rollback query
+}
+
+// Run the Migrator with migration queries provided by the Loader.
+func Run(ctx context.Context, m Migrator, l Loader) error {
+	ms, err := loadMigrations(l.Load())
+	if err != nil {
+		return err
+	}
+	return runMigration(ctx, m, ms)
+}
+
+func loadMigrations(ms []*Migration, err error) ([]*Migration, error) {
+	if err != nil {
+		return nil, err
+	}
+
+	sort.SliceStable(ms, func(i, j int) bool {
+		return ms[i].ID < ms[j].ID
+	})
+
+	for i, m := range ms {
+		switch want := i + 1; {
+		case m.ID < want:
+			return nil, fmt.Errorf("duplicate migration number: %d (%s)", m.ID, m.Name)
+		case m.ID > want:
+			return nil, fmt.Errorf("missing migration number: %d (have %d)", want, m.ID)
+		default:
+			// pass
+		}
+	}
+	return ms, nil
+}
+
+func runMigration(ctx context.Context, m Migrator, ms []*Migration) error {
+	if err := m.Lock(ctx); err != nil {
+		return err
+	}
+
+	var err error
+	defer func() {
+		if errUnlock := m.Unlock(ctx); err == nil && errUnlock != nil {
+			err = errUnlock
+		}
+	}()
+
+	err = runLockedMigration(ctx, m, ms)
+	return err
+}
+
+func runLockedMigration(ctx context.Context, m Migrator, ms []*Migration) error {
+	currentVersion, err := m.Version(ctx)
+	if err != nil {
+		return err
+	}
+
+	// TODO: configure
+	targetVersion := len(ms)
+	switch {
+	case targetVersion < 0 || len(ms) < targetVersion:
+		fallthrough
+	case currentVersion < 0 || len(ms) < currentVersion:
+		return fmt.Errorf("target version %d is outside of range 0..%d ", targetVersion, len(ms))
+	}
+
+	direction := 1
+	if currentVersion > targetVersion {
+		direction = -1
+	}
+
+	for currentVersion != targetVersion {
+		current := ms[currentVersion]
+		sequence := current.ID
+		query := current.Apply
+
+		if direction == -1 {
+			current = ms[currentVersion-1]
+			sequence = current.ID - 1
+			query = current.Rollback
+		}
+
+		if err := m.Exec(ctx, query); err != nil {
+			return err
+		}
+
+		if err := m.SetVersion(ctx, sequence); err != nil {
+			return err
+		}
+		currentVersion += direction
+	}
+	return nil
+}