go/migrate/migrate_test.go

package migrate_test

import (
	"bytes"
	"context"
	"crypto/rand"
	"database/sql"
	"embed"
	"errors"
	"io"
	"io/fs"
	"log/slog"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/stdlib"
	"github.com/pressly/goose/v3"
	_ "modernc.org/sqlite"

	"go.auvem.com/corestack/migrate"
)

// invariant: embedded fixtures are read-only; no test mutates shared migration state.
//
//go:embed testdata
var fixtures embed.FS

func TestSQLite(t *testing.T) {
	for _, test := range []struct {
		name string
		run  func(*testing.T, *sql.DB, goose.Dialect)
	}{
		{"flow", testFlow}, {"sql-only", testSQLOnly}, {"failure", testFailure},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "migrate.db"))
			if err != nil {
				t.Fatal(err)
			}
			t.Cleanup(func() {
				if err := db.Close(); err != nil {
					t.Error(err)
				}
			})
			test.run(t, db, goose.DialectSQLite3)
		})
	}
}

func TestPostgres(t *testing.T) {
	for _, test := range []struct {
		name string
		run  func(*testing.T, *sql.DB, goose.Dialect)
	}{
		{"flow", testFlow}, {"sql-only", testSQLOnly}, {"failure", testFailure},
	} {
		t.Run(test.name, func(t *testing.T) { test.run(t, postgresDatabase(t)(), goose.DialectPostgres) })
	}
}

func testFlow(t *testing.T, db *sql.DB, dialect goose.Dialect) {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
	defer cancel()
	var logs bytes.Buffer
	goMigration := goose.NewGoMigration(20261027120000,
		&goose.GoFunc{RunTx: func(ctx context.Context, tx *sql.Tx) error {
			_, err := tx.ExecContext(ctx, "INSERT INTO migration_events (version) VALUES (20261027120000)")
			return err
		}},
		&goose.GoFunc{RunTx: func(ctx context.Context, tx *sql.Tx) error {
			_, err := tx.ExecContext(ctx, "DELETE FROM migration_events WHERE version = 20261027120000")
			return err
		}})
	p, err := migrate.New(db, dialect, fixtures, "testdata/migrations", slog.New(slog.NewJSONHandler(&logs, nil)), goMigration)
	if err != nil {
		t.Fatal(err)
	}
	versions := []int64{7, 20261026093000, 20261027114500, 20261027120000}
	// Registering on one provider must not leak into another provider in the same process.
	other := newProvider(t, db, dialect, "testdata/migrations")
	if len(other.ListSources()) != 3 {
		t.Fatal("Go migration leaked to another provider")
	}
	assertState(t, ctx, p, 0, versions, 0)
	results, err := p.UpTo(ctx, versions[1])
	if err != nil {
		t.Fatal(err)
	}
	assertResults(t, results, versions[:2])
	assertState(t, ctx, p, versions[1], versions, 2)
	results, err = p.Up(ctx)
	if err != nil {
		t.Fatal(err)
	}
	assertResults(t, results, versions[2:])
	assertState(t, ctx, p, versions[3], versions, 4)
	assertEvents(t, ctx, db, versions)
	results, err = p.Up(ctx)
	if err != nil || len(results) != 0 {
		t.Fatalf("idempotent Up = %v, %v", results, err)
	}
	results, err = p.DownTo(ctx, versions[1])
	if err != nil {
		t.Fatal(err)
	}
	assertResults(t, results, []int64{versions[3], versions[2]})
	assertState(t, ctx, p, versions[1], versions, 2)
	assertEvents(t, ctx, db, versions[:2])
	results, err = p.DownTo(ctx, versions[0])
	if err != nil {
		t.Fatal(err)
	}
	assertResults(t, results, []int64{versions[1]})
	assertState(t, ctx, p, versions[0], versions, 1)
	assertEvents(t, ctx, db, versions[:1])
	if _, err := p.DownTo(ctx, 0); err != nil {
		t.Fatal(err)
	}
	assertState(t, ctx, p, 0, versions, 0)
	results, err = p.Up(ctx)
	if err != nil {
		t.Fatal(err)
	}
	assertResults(t, results, versions)
	assertEvents(t, ctx, db, versions)
	if !strings.Contains(logs.String(), `"logger":"goose"`) || !strings.Contains(logs.String(), `"version":20261027114500`) {
		t.Fatalf("missing structured migration logs: %s", logs.String())
	}
}

func testSQLOnly(t *testing.T, db *sql.DB, dialect goose.Dialect) {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
	defer cancel()
	for _, test := range []struct {
		dir      string
		versions []int64
	}{
		{"testdata/numeric", []int64{7, 10}},
		{"testdata/migrations", []int64{7, 20261026093000, 20261027114500}},
	} {
		p := newProvider(t, db, dialect, test.dir)
		assertState(t, ctx, p, 0, test.versions, 0)
		results, err := p.UpTo(ctx, test.versions[1]-1)
		if err != nil {
			t.Fatal(err)
		}
		assertResults(t, results, test.versions[:1])
		assertState(t, ctx, p, test.versions[0], test.versions, 1)
		results, err = p.Up(ctx)
		if err != nil {
			t.Fatal(err)
		}
		assertResults(t, results, test.versions[1:])
		assertState(t, ctx, p, test.versions[len(test.versions)-1], test.versions, len(test.versions))
		assertEvents(t, ctx, db, test.versions)
		results, err = p.DownTo(ctx, test.versions[1]-1)
		if err != nil {
			t.Fatal(err)
		}
		wantDown := slices.Clone(test.versions[1:])
		slices.Reverse(wantDown)
		assertResults(t, results, wantDown)
		assertState(t, ctx, p, test.versions[0], test.versions, 1)
		if _, err := p.DownTo(ctx, 0); err != nil {
			t.Fatal(err)
		}
		assertState(t, ctx, p, 0, test.versions, 0)
	}
}

func testFailure(t *testing.T, db *sql.DB, dialect goose.Dialect) {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
	defer cancel()
	p := newProvider(t, db, dialect, "testdata/failure")
	_, err := p.Up(ctx)
	var partial *goose.PartialError
	if !errors.As(err, &partial) || partial.Failed.Source.Version != 20261027114500 ||
		!strings.Contains(err.Error(), "missing_migration_table") || !strings.Contains(err.Error(), "20261027114500") {
		t.Fatalf("expected clear failed timestamp migration error, got %v", err)
	}
	assertResults(t, partial.Applied, []int64{7, 20261026093000})
	assertState(t, ctx, p, 20261026093000, []int64{7, 20261026093000, 20261027114500}, 2)
	assertEvents(t, ctx, db, []int64{7, 20261026093000})
	// A second attempt must fail at the same version without repeating committed work.
	if _, err := p.Up(ctx); err == nil {
		t.Fatal("retry unexpectedly succeeded")
	}
	assertEvents(t, ctx, db, []int64{7, 20261026093000})
}

func newProvider(t *testing.T, db *sql.DB, dialect goose.Dialect, dir string, migrations ...*goose.Migration) *goose.Provider {
	t.Helper()
	p, err := migrate.New(db, dialect, fixtures, dir, slog.New(slog.NewTextHandler(io.Discard, nil)), migrations...)
	if err != nil {
		t.Fatal(err)
	}
	return p
}

func assertState(t *testing.T, ctx context.Context, p *goose.Provider, current int64, versions []int64, applied int) {
	t.Helper()
	var sources []int64
	for _, source := range p.ListSources() {
		sources = append(sources, source.Version)
	}
	if !slices.Equal(sources, versions) {
		t.Fatalf("ListSources = %v; want %v", sources, versions)
	}
	got, err := p.GetDBVersion(ctx)
	if err != nil || got != current {
		t.Fatalf("GetDBVersion = %d, %v; want %d", got, err, current)
	}
	got, latest, err := p.GetVersions(ctx)
	if err != nil || got != current || latest != versions[len(versions)-1] {
		t.Fatalf("GetVersions = %d, %d, %v", got, latest, err)
	}
	pending, err := p.HasPending(ctx)
	if err != nil || pending != (applied < len(versions)) {
		t.Fatalf("HasPending = %v, %v", pending, err)
	}
	status, err := p.Status(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(status) != len(versions) {
		t.Fatalf("Status length = %d; want %d", len(status), len(versions))
	}
	for i, s := range status {
		want := goose.StatePending
		if i < applied {
			want = goose.StateApplied
		}
		if s.Source.Version != versions[i] || s.State != want {
			t.Fatalf("Status[%d] = %+v, version %d; want %s, %d", i, s, s.Source.Version, want, versions[i])
		}
	}
}

func assertResults(t *testing.T, results []*goose.MigrationResult, want []int64) {
	t.Helper()
	got := make([]int64, 0, len(results))
	for _, r := range results {
		got = append(got, r.Source.Version)
	}
	if !slices.Equal(got, want) {
		t.Fatalf("applied versions = %v; want %v", got, want)
	}
}

func assertEvents(t *testing.T, ctx context.Context, db *sql.DB, want []int64) {
	t.Helper()
	rows, err := db.QueryContext(ctx, "SELECT version FROM migration_events ORDER BY version")
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	var got []int64
	for rows.Next() {
		var v int64
		if err := rows.Scan(&v); err != nil {
			t.Fatal(err)
		}
		got = append(got, v)
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(got, want) {
		t.Fatalf("events = %v; want %v", got, want)
	}
}

// postgresDatabase owns only its random schema, never the supplied database.
func postgresDatabase(t *testing.T) func() *sql.DB {
	t.Helper()
	dsn := os.Getenv("MIGRATE_POSTGRES_DSN")
	if dsn == "" {
		if os.Getenv("CI") != "" {
			t.Fatal("MIGRATE_POSTGRES_DSN is required in CI")
		}
		t.Skip("set MIGRATE_POSTGRES_DSN to run Postgres integration tests")
	}
	config, err := pgx.ParseConfig(dsn)
	if err != nil {
		t.Fatal(err)
	}
	admin := stdlib.OpenDB(*config)
	t.Cleanup(func() {
		if err := admin.Close(); err != nil {
			t.Error(err)
		}
	})
	schema := "migrate_" + strings.ToLower(rand.Text())
	ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
	defer cancel()
	if _, err := admin.ExecContext(ctx, "CREATE SCHEMA "+schema); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
		defer cancel()
		if _, err := admin.ExecContext(ctx, "DROP SCHEMA "+schema+" CASCADE"); err != nil {
			t.Error(err)
		}
	})
	config.RuntimeParams["search_path"] = schema
	return func() *sql.DB {
		db := stdlib.OpenDB(*config)
		t.Cleanup(func() {
			if err := db.Close(); err != nil {
				t.Error(err)
			}
		})
		return db
	}
}

func TestPostgresConcurrentUp(t *testing.T) {
	open := postgresDatabase(t)
	ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
	defer cancel()
	const workers = 4
	providers := make([]*goose.Provider, workers)
	for i := range providers {
		// Each provider owns a distinct pool and Go migration instance, like separate daemons.
		pause := goose.NewGoMigration(3, &goose.GoFunc{RunTx: func(ctx context.Context, tx *sql.Tx) error {
			_, err := tx.ExecContext(ctx, "SELECT pg_sleep(0.2)")
			return err
		}}, nil)
		providers[i] = newProvider(t, open(), goose.DialectPostgres, "testdata/migrations", pause)
	}
	type outcome struct {
		results []*goose.MigrationResult
		err     error
	}
	start := make(chan struct{})
	done := make(chan outcome, workers)
	for _, p := range providers {
		go func() {
			select {
			case <-start:
			case <-ctx.Done():
				done <- outcome{err: ctx.Err()}
				return
			}
			results, err := p.Up(ctx)
			done <- outcome{results, err}
		}()
	}
	close(start)
	var applied []int64
	for range workers {
		result := <-done
		if result.err != nil {
			t.Error(result.err)
		}
		for _, r := range result.results {
			applied = append(applied, r.Source.Version)
		}
	}
	slices.Sort(applied)
	if want := []int64{3, 7, 20261026093000, 20261027114500}; !slices.Equal(applied, want) {
		t.Fatalf("concurrent Up applied = %v; want exactly once each: %v", applied, want)
	}
	assertEvents(t, ctx, open(), []int64{7, 20261026093000, 20261027114500})
	for _, p := range providers {
		assertState(t, ctx, p, 20261027114500, applied, len(applied))
	}
}

func TestConstructor(t *testing.T) {
	db, err := sql.Open("sqlite", ":memory:")
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	for _, test := range []struct {
		name       string
		db         *sql.DB
		dialect    goose.Dialect
		dir        string
		logger     *slog.Logger
		migrations []*goose.Migration
		contains   string
	}{
		{name: "nil database", dialect: goose.DialectSQLite3, dir: "testdata/migrations", logger: logger, contains: "db must not be nil"},
		{name: "nil logger", db: db, dialect: goose.DialectSQLite3, dir: "testdata/migrations", contains: "slog logger must not be nil"},
		{name: "invalid directory", db: db, dialect: goose.DialectSQLite3, dir: "../migrations", logger: logger, contains: "directory"},
		{name: "missing directory", db: db, dialect: goose.DialectSQLite3, dir: "missing", logger: logger, contains: "migrate:"},
		{name: "invalid dialect", db: db, dialect: "unknown", dir: "testdata/migrations", logger: logger, contains: "dialect"},
		{name: "nil migration", db: db, dialect: goose.DialectSQLite3, dir: "testdata/migrations", logger: logger, migrations: []*goose.Migration{nil}, contains: "must not be nil"},
		{name: "duplicate version", db: db, dialect: goose.DialectSQLite3, dir: "testdata/migrations", logger: logger, migrations: []*goose.Migration{goose.NewGoMigration(7, nil, nil)}, contains: "7"},
	} {
		t.Run(test.name, func(t *testing.T) {
			p, err := migrate.New(test.db, test.dialect, fixtures, test.dir, test.logger, test.migrations...)
			if err == nil || p != nil || !strings.Contains(err.Error(), test.contains) {
				t.Fatalf("New = %v, %v; want error containing %q", p, err, test.contains)
			}
		})
	}
	// Construction and validation are offline, even when the database is closed.
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	if _, err := migrate.New(db, goose.DialectSQLite3, fixtures, "testdata/migrations", logger); err != nil {
		t.Fatal(err)
	}
}

func TestConstructorDirectory(t *testing.T) {
	db, err := sql.Open("sqlite", ":memory:")
	if err != nil {
		t.Fatal(err)
	}
	// Directory validation and Go-only construction must remain offline.
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	for _, test := range []struct {
		dir     string
		wantErr string
	}{
		{dir: "missing", wantErr: "directory"},
		{dir: "testdata/migrations/00007_events.sql", wantErr: "not a directory"},
		{dir: "."},
		{dir: "testdata"},
	} {
		t.Run(test.dir, func(t *testing.T) {
			p, err := migrate.New(db, goose.DialectSQLite3, fixtures, test.dir, logger, goose.NewGoMigration(10, nil, nil))
			if test.wantErr != "" {
				if err == nil || p != nil || !strings.Contains(err.Error(), test.wantErr) || !strings.Contains(err.Error(), test.dir) {
					t.Fatalf("New = %v, %v; want directory error containing %q and %q", p, err, test.wantErr, test.dir)
				}
				if test.dir == "missing" && !errors.Is(err, fs.ErrNotExist) {
					t.Fatalf("New error = %v; want fs.ErrNotExist", err)
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if sources := p.ListSources(); len(sources) != 1 || sources[0].Version != 10 {
				t.Fatalf("ListSources = %v; want Go migration 10 only", sources)
			}
		})
	}
}