gomigr.go

// Implementation of the public library facade: the Migrator type with Up,
// Down, Status, Close methods and the New constructor. The cmd/gomigr CLI
// calls these methods directly rather than duplicating the migration logic.
package gomigr

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/stdlib"

	"github.com/therox/gomigr/internal/runner"
	"github.com/therox/gomigr/internal/store"
	"github.com/therox/gomigr/parser"
)

// ErrLockBusy indicates that another gomigr instance currently holds the
// advisory lock against this database. Re-exported from internal/store so
// library consumers can match it via errors.Is without depending on the
// internal package.
var ErrLockBusy = store.ErrLockBusy

// ErrDownScopeRequired is returned from Migrator.Down when DownOptions sets
// none of To, Steps, All — the library refuses to guess what the caller
// meant. Previously an "empty toVersion" was treated as "revert everything";
// now that is an explicit opt-in through DownOptions.All.
var ErrDownScopeRequired = errors.New("gomigr: DownOptions requires To, Steps or All")

// UpOptions configures Migrator.Up.
type UpOptions struct {
	// To is the upper version bound (inclusive). An empty string means no
	// upper bound.
	To string
	// Steps caps the number of pending migrations to apply (not counting
	// transitive dependencies pulled in for integrity). 0 means no cap.
	// A negative value is a validation error.
	Steps int
}

// DownOptions configures Migrator.Down. Exactly one of the fields must be
// set; otherwise ErrDownScopeRequired is returned.
type DownOptions struct {
	// To is the version to roll back to (applied migrations with
	// version > To get reverted).
	To string
	// Steps reverts the N most recent applied migrations by id; must be > 0.
	Steps int
	// All is an explicit request to revert every applied migration. It is
	// the equivalent of the old Down(ctx, "") behaviour but requires
	// explicit opt-in. The CLI layers interactive confirmation on top.
	All bool
}

// Options is the constructor argument for New. The caller fills the fields
// (the CLI does it after loading YAML/ENV/flags; external Go callers do it
// programmatically). The library itself does not read environment variables.
type Options struct {
	// DSN is the required PostgreSQL connection string. Both URL form
	// (postgres://...) and pgx keyword form (host=... user=...) are
	// supported.
	DSN string
	// MigrationsDir is the required path to the directory holding *.sql
	// migrations.
	MigrationsDir string
	// User is the value written into the applied_by column. An empty
	// string is resolved inside the library through os/user.Current()
	// and finally to "unknown".
	User string
	// Logger is the *slog.Logger used for all messages. nil falls back to
	// a discard handler.
	Logger *slog.Logger
}

// MigrationStatus is one row of Migrator.Status() output.
type MigrationStatus struct {
	ID        int64
	Version   string
	Name      string
	Status    string
	AppliedBy string
	AppliedAt time.Time
}

// Migrator is the public library facade. It is created with New and must
// always be released with Close.
type Migrator struct {
	opts   Options
	db     *sql.DB
	store  *store.Store
	runner *runner.Runner
	logger *slog.Logger
}

// New validates Options, opens the internal *sql.DB through pgx/stdlib in
// simple-protocol mode (so multi-statement bodies in Up/Down SQL work),
// pings the database and creates schema_migrations if it is missing.
//
// The pool is intentionally tiny: SetMaxOpenConns(2), SetMaxIdleConns(1).
// Migrations themselves do not need parallelism, but a second slot is
// required by the advisory lock, which holds one connection for the entire
// duration of a batch.
func New(ctx context.Context, opts Options) (*Migrator, error) {
	if opts.DSN == "" {
		return nil, errors.New("gomigr: Options.DSN is required")
	}
	if opts.MigrationsDir == "" {
		return nil, errors.New("gomigr: Options.MigrationsDir is required")
	}
	logger := opts.Logger
	if logger == nil {
		logger = slog.New(slog.NewTextHandler(io.Discard, nil))
	}

	cfg, err := pgx.ParseConfig(opts.DSN)
	if err != nil {
		return nil, fmt.Errorf("gomigr: parse dsn: %w", err)
	}
	// Simple protocol lets the driver execute multi-statement bodies in one
	// ExecContext call. Without it pgx uses the extended protocol, in which
	// several queries on a single line raise an error.
	cfg.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol
	// GetConnector + sql.OpenDB does not leave entries in the process-global
	// map of pgx-stdlib (unlike RegisterConnConfig + sql.Open("pgx", connStr)),
	// so repeated New calls from a long-lived application do not leak.
	db := sql.OpenDB(stdlib.GetConnector(*cfg))
	db.SetMaxOpenConns(2)
	db.SetMaxIdleConns(1)
	if err := db.PingContext(ctx); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("gomigr: ping: %w", err)
	}
	s := store.New(db)
	if err := s.EnsureSchema(ctx); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("gomigr: %w", err)
	}
	return &Migrator{
		opts:   opts,
		db:     db,
		store:  s,
		runner: runner.New(s, logger, opts.User),
		logger: logger,
	}, nil
}

// Up applies pending migrations according to UpOptions. A zero UpOptions is
// equivalent to "apply everything pending with no upper bound". The
// advisory lock is held for the entire run. Returns ErrLockBusy if the lock
// is busy.
//
// Validation: Steps < 0 fails before connecting to the database. Steps == 0
// is treated as "no cap" — symmetric with an empty To.
func (m *Migrator) Up(ctx context.Context, opts UpOptions) error {
	if opts.Steps < 0 {
		return errors.New("gomigr: UpOptions.Steps must be >= 0")
	}
	return m.runBatch(ctx, func(all []*parser.Migration) error {
		return m.runner.Up(ctx, all, runner.UpOptions{To: opts.To, Steps: opts.Steps})
	})
}

// Down reverts applied migrations according to DownOptions. Exactly one of
// To, Steps, All must be set (To non-empty, Steps > 0, or All == true);
// otherwise ErrDownScopeRequired is returned. Setting two or more is a
// mutual-exclusion error. The advisory lock is held for the entire run.
func (m *Migrator) Down(ctx context.Context, opts DownOptions) error {
	if opts.Steps < 0 {
		return errors.New("gomigr: DownOptions.Steps must be >= 0")
	}
	set := 0
	if opts.To != "" {
		set++
	}
	if opts.Steps > 0 {
		set++
	}
	if opts.All {
		set++
	}
	switch set {
	case 0:
		return ErrDownScopeRequired
	case 1:
		// ok
	default:
		return errors.New("gomigr: DownOptions fields To, Steps, All are mutually exclusive")
	}
	return m.runBatch(ctx, func(all []*parser.Migration) error {
		return m.runner.Down(ctx, all, runner.DownOptions{
			To:    opts.To,
			Steps: opts.Steps,
			All:   opts.All,
		})
	})
}

// Status returns every applied migration (including rows flagged as
// checksum_mismatch), ordered by id — the chronological apply order
// assigned by the BIGSERIAL column on INSERT into schema_migrations.
func (m *Migrator) Status(ctx context.Context) ([]MigrationStatus, error) {
	rows, err := m.store.ListApplied(ctx)
	if err != nil {
		return nil, err
	}
	out := make([]MigrationStatus, 0, len(rows))
	for _, r := range rows {
		out = append(out, MigrationStatus{
			ID:        r.ID,
			Version:   r.Version,
			Name:      r.Name,
			Status:    r.Status,
			AppliedBy: r.AppliedBy,
			AppliedAt: r.AppliedAt,
		})
	}
	return out, nil
}

// Close closes the internal *sql.DB. The Migrator must not be used after
// Close. Calling Close more than once is safe.
func (m *Migrator) Close() error {
	if m == nil || m.db == nil {
		return nil
	}
	db := m.db
	m.db = nil
	return db.Close()
}

// runBatch is the shared wrapper around Up/Down: load migrations from disk,
// acquire the advisory lock for the whole operation, then call the provided
// function while the lock is held.
func (m *Migrator) runBatch(ctx context.Context, fn func([]*parser.Migration) error) error {
	all, err := parser.LoadDir(m.opts.MigrationsDir)
	if err != nil {
		return err
	}
	for _, mig := range all {
		if mig.DownSQL == "" {
			// An empty Down section is allowed by the spec ("warning in
			// the log, not an error"). Emit through Options.Logger so we
			// do not depend on the global slog.Default.
			m.logger.Warn("empty Down section", "version", mig.Version, "name", mig.Name)
		}
	}
	lock, err := m.store.AcquireLock(ctx)
	if err != nil {
		return err
	}
	defer func() {
		// Release the advisory lock on a non-cancellable context: even if
		// the caller's ctx is already cancelled (Ctrl-C / timeout) we
		// still want pg_advisory_unlock to run and the connection to
		// return to the pool.
		releaseCtx := context.WithoutCancel(ctx)
		if rerr := lock.Release(releaseCtx); rerr != nil {
			m.logger.Error("release advisory lock", "err", rerr)
		}
	}()
	return fn(all)
}