rivermigrate

Documentation

Overview

Package rivermigrate provides a Go API for running migrations as alternative to migrating via the bundled CLI.

Example (Migrate)

Example_migrate demonstrates the use of River's Go migration API by migrating up and down.

package main

import (
	"context"
	"fmt"
	"strings"

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

	"github.com/riverqueue/river/internal/riverinternaltest"
	"github.com/riverqueue/river/riverdriver"
	"github.com/riverqueue/river/riverdriver/riverpgxv5"
	"github.com/riverqueue/river/rivermigrate"
)

// Example_migrate demonstrates the use of River's Go migration API by migrating
// up and down.
func main() {
	ctx := context.Background()

	// Use a dedicated Postgres schema for this example so we can migrate and drop it at will:
	schemaName := "migration_example"
	poolConfig := riverinternaltest.DatabaseConfig("river_test_example")
	poolConfig.ConnConfig.RuntimeParams["search_path"] = schemaName

	dbPool, err := pgxpool.NewWithConfig(ctx, poolConfig)
	if err != nil {
		panic(err)
	}
	defer dbPool.Close()

	driver := riverpgxv5.New(dbPool)
	migrator, err := rivermigrate.New(driver, nil)
	if err != nil {
		panic(err)
	}

	// Create the schema used for this example. Drop it when we're done.
	// This isn't necessary outside this test.
	if _, err := dbPool.Exec(ctx, "CREATE SCHEMA IF NOT EXISTS "+schemaName); err != nil {
		panic(err)
	}
	defer dropRiverSchema(ctx, driver, schemaName)

	printVersions := func(res *rivermigrate.MigrateResult) {
		for _, version := range res.Versions {
			fmt.Printf("Migrated [%s] version %d\n", strings.ToUpper(string(res.Direction)), version.Version)
		}
	}

	// Migrate to version 3. An actual call may want to omit all MigrateOpts,
	// which will default to applying all available up migrations.
	res, err := migrator.Migrate(ctx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{
		TargetVersion: 3,
	})
	if err != nil {
		panic(err)
	}
	printVersions(res)

	// Migrate down by three steps. Down migrating defaults to running only one
	// step unless overridden by an option like MaxSteps or TargetVersion.
	res, err = migrator.Migrate(ctx, rivermigrate.DirectionDown, &rivermigrate.MigrateOpts{
		MaxSteps: 3,
	})
	if err != nil {
		panic(err)
	}
	printVersions(res)

}

func dropRiverSchema[TTx any](ctx context.Context, driver riverdriver.Driver[TTx], schemaName string) {
	_, err := driver.GetExecutor().Exec(ctx, "DROP SCHEMA IF EXISTS "+schemaName+" CASCADE;")
	if err != nil {
		panic(err)
	}
}

Output:

Migrated [UP] version 1
Migrated [UP] version 2
Migrated [UP] version 3
Migrated [DOWN] version 3
Migrated [DOWN] version 2
Migrated [DOWN] version 1

Example (MigrateDatabaseSQL)

Example_migrateDatabaseSQL demonstrates the use of River's Go migration API through Go's built-in database/sql package.

ctx := context.Background()

// Use a dedicated Postgres schema for this example so we can migrate and drop it at will:
schemaName := "migration_example_dbsql"
url := riverinternaltest.DatabaseURL("river_test_example") + "&search_path=" + schemaName
dbPool, err := sql.Open("pgx", url)
if err != nil {
	panic(err)
}
defer dbPool.Close()

driver := riverdatabasesql.New(dbPool)
migrator, err := rivermigrate.New(driver, nil)
if err != nil {
	panic(err)
}

// Create the schema used for this example. Drop it when we're done.
// This isn't necessary outside this test.
if _, err := dbPool.ExecContext(ctx, "CREATE SCHEMA IF NOT EXISTS "+schemaName); err != nil {
	panic(err)
}
defer dropRiverSchema(ctx, driver, schemaName)

printVersions := func(res *rivermigrate.MigrateResult) {
	for _, version := range res.Versions {
		fmt.Printf("Migrated [%s] version %d\n", strings.ToUpper(string(res.Direction)), version.Version)
	}
}

// Migrate to version 3. An actual call may want to omit all MigrateOpts,
// which will default to applying all available up migrations.
res, err := migrator.Migrate(ctx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{
	TargetVersion: 3,
})
if err != nil {
	panic(err)
}
printVersions(res)

// Migrate down by three steps. Down migrating defaults to running only one
// step unless overridden by an option like MaxSteps or TargetVersion.
res, err = migrator.Migrate(ctx, rivermigrate.DirectionDown, &rivermigrate.MigrateOpts{
	MaxSteps: 3,
})
if err != nil {
	panic(err)
}
printVersions(res)

Output:

Migrated [UP] version 1
Migrated [UP] version 2
Migrated [UP] version 3
Migrated [DOWN] version 3
Migrated [DOWN] version 2
Migrated [DOWN] version 1

Index

• type Config
• type Direction
• type MigrateOpts
• type MigrateResult
• type MigrateVersion
• type Migration
• type Migrator
  • func New[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Migrator[TTx], error)
  • func (m *Migrator[TTx]) AllVersions() []Migration
  • func (m *Migrator[TTx]) ExistingVersions(ctx context.Context) ([]Migration, error)
  • func (m *Migrator[TTx]) ExistingVersionsTx(ctx context.Context, tx TTx) ([]Migration, error)
  • func (m *Migrator[TTx]) GetVersion(version int) (Migration, error)
  • func (m *Migrator[TTx]) Migrate(ctx context.Context, direction Direction, opts *MigrateOpts) (*MigrateResult, error)
  • func (m *Migrator[TTx]) MigrateTx(ctx context.Context, tx TTx, direction Direction, opts *MigrateOpts) (*MigrateResult, error)
  • func (m *Migrator[TTx]) Validate(ctx context.Context) (*ValidateResult, error)
  • func (m *Migrator[TTx]) ValidateTx(ctx context.Context, tx TTx) (*ValidateResult, error)
• type ValidateResult

Examples

• Package (Migrate)
• Package (MigrateDatabaseSQL)

Types

type Config

type Config struct {
	// Line is the migration line to use. Most drivers will only have a single
	// line, which is `main`.
	//
	// Defaults to `main`.
	Line string

	// Logger is the structured logger to use for logging purposes. If none is
	// specified, logs will be emitted to STDOUT with messages at warn level
	// or higher.
	Logger *slog.Logger
}

Config contains configuration for Migrator.

type Direction

type Direction string

const (
	DirectionDown Direction = "down"
	DirectionUp   Direction = "up"
)

type MigrateOpts

type MigrateOpts struct {
	DryRun bool

	// MaxSteps is the maximum number of migrations to apply either up or down.
	// When migrating in the up direction, migrates an unlimited number of steps
	// by default. When migrating in the down direction, migrates only a single
	// step by default (set TargetVersion to -1 to apply unlimited steps down).
	// Set to -1 to apply no migrations (for testing/checking purposes).
	MaxSteps int

	// TargetVersion is a specific migration version to apply migrations to. The
	// version must exist and it must be in the possible list of migrations to
	// apply. e.g. If requesting an up migration with version 3, version 3 must
	// not already be applied.
	//
	// When applying migrations up, migrations are applied including the target
	// version, so when starting at version 0 and requesting version 3, versions
	// 1, 2, and 3 would be applied. When applying migrations down, down
	// migrations are applied excluding the target version, so when starting at
	// version 5 and requesting version 3, down migrations for versions 5 and 4
	// would be applied, leaving the final schema at version 3.
	//
	// When migrating down, TargetVersion can be set to the special value of -1
	// to apply all down migrations (i.e. River schema is removed completely).
	TargetVersion int
}

MigrateOpts are options for a migrate operation.

type MigrateResult

type MigrateResult struct {
	// Direction is the direction that migration occurred (up or down).
	Direction Direction

	// Versions are migration versions that were added (for up migrations) or
	// removed (for down migrations) for this run.
	Versions []MigrateVersion
}

MigrateResult is the result of a migrate operation.

type MigrateVersion

type MigrateVersion struct {
	// Duration is the amount of time it took to apply the migration.
	Duration time.Duration

	// Name is a human-friendly name for the migration derived from its
	// filename.
	Name string

	// SQL is the SQL that was applied along with the migration.
	SQL string

	// Version is the version of the migration applied.
	Version int
}

MigrateVersion is the result for a single applied migration.

type Migration

type Migration struct {
	// Name is a human-friendly name for the migration derived from its
	// filename.
	Name string

	// SQLDown is the s SQL for the migration's down direction.
	SQLDown string

	// SQLUp is the s SQL for the migration's up direction.
	SQLUp string

	// Version is the integer version number of this migration.
	Version int
}

Migration is a bundled migration containing a version (e.g. 1, 2, 3), and SQL for up and down directions.

type Migrator

type Migrator[TTx any] struct {
	baseservice.BaseService
	// contains filtered or unexported fields
}

Migrator is a database migration tool for River which can run up or down migrations in order to establish the schema that the queue needs to run.

func New

func New[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Migrator[TTx], error)

New returns a new migrator with the given database driver and configuration. The config parameter may be omitted as nil.

Two drivers are supported for migrations, one for Pgx v5 and one for the built-in database/sql package for use with migration frameworks like Goose. See packages riverpgxv5 and riverdatabasesql respectively.

The function takes a generic parameter TTx representing a transaction type, but it can be omitted because it'll generally always be inferred from the driver. For example:

import "github.com/riverqueue/river/riverdriver/riverpgxv5"
import "github.com/riverqueue/rivermigrate"

...

dbPool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
	// handle error
}
defer dbPool.Close()

migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
if err != nil {
	// handle error
}

func (*Migrator[TTx]) AllVersions

func (m *Migrator[TTx]) AllVersions() []Migration

AllVersions gets information on all known migration versions.

func (*Migrator[TTx]) ExistingVersions

func (m *Migrator[TTx]) ExistingVersions(ctx context.Context) ([]Migration, error)

ExistingVersions gets the existing set of versions that have been migrated in the database, ordered by version.

func (*Migrator[TTx]) ExistingVersionsTx

func (m *Migrator[TTx]) ExistingVersionsTx(ctx context.Context, tx TTx) ([]Migration, error)

ExistingVersions gets the existing set of versions that have been migrated in the database, ordered by version.

This variant checks for existing versions in a transaction.

func (*Migrator[TTx]) GetVersion

func (m *Migrator[TTx]) GetVersion(version int) (Migration, error)

GetVersion gets information about a specific migration version. An error is returned if a versions is requested that doesn't exist.

func (*Migrator[TTx]) Migrate

func (m *Migrator[TTx]) Migrate(ctx context.Context, direction Direction, opts *MigrateOpts) (*MigrateResult, error)

Migrate migrates the database in the given direction (up or down). The opts parameter may be omitted for convenience.

By default, applies all outstanding migrations when moving in the up direction, but for safety, only one step when moving in the down direction. To migrate more than one step down, MigrateOpts.MaxSteps or MigrateOpts.TargetVersion are available. Setting MigrateOpts.TargetVersion to -1 will apply every available downstep so that River's schema is removed completely.

res, err := migrator.Migrate(ctx, rivermigrate.DirectionUp, nil)
if err != nil {
	// handle error
}

func (*Migrator[TTx]) MigrateTx

func (m *Migrator[TTx]) MigrateTx(ctx context.Context, tx TTx, direction Direction, opts *MigrateOpts) (*MigrateResult, error)

Migrate migrates the database in the given direction (up or down). The opts parameter may be omitted for convenience.

By default, applies all outstanding migrations when moving in the up direction, but for safety, only one step when moving in the down direction. To migrate more than one step down, MigrateOpts.MaxSteps or MigrateOpts.TargetVersion are available. Setting MigrateOpts.TargetVersion to -1 will apply every available downstep so that River's schema is removed completely.

res, err := migrator.MigrateTx(ctx, tx, rivermigrate.DirectionUp, nil)
if err != nil {
	// handle error
}

This variant lets a caller run migrations within a transaction. Postgres DDL is transactional, so migration changes aren't visible until the transaction commits, and are rolled back if the transaction rolls back.

Deprecated: Use Migrate instead. Certain migrations cannot be batched together in a single transaction, so this method is not recommended.

func (*Migrator[TTx]) Validate

func (m *Migrator[TTx]) Validate(ctx context.Context) (*ValidateResult, error)

Validate validates the current state of migrations, returning an unsuccessful validation and usable message in case there are migrations that haven't yet been applied.

func (*Migrator[TTx]) ValidateTx

func (m *Migrator[TTx]) ValidateTx(ctx context.Context, tx TTx) (*ValidateResult, error)

Validate validates the current state of migrations, returning an unsuccessful validation and usable message in case there are migrations that haven't yet been applied.

This variant lets a caller validate within a transaction.

type ValidateResult

type ValidateResult struct {
	// Messages contain informational messages of what wasn't valid in case of a
	// failed validation. Always empty if OK is true.
	Messages []string

	// OK is true if validation completed with no problems.
	OK bool
}

ValidateResult is the result of a validation operation.