Better handle non-existent migration line + suggestions with Levensht…

…ein distance

This is related to #546. That was a documentation problem, but there are
some code problems that are related.

The first is that an unknown migration line is a panic in the CLI. This
is a byproduct of the fact that `rivermigrate.New` checks the migration
line name, but doesn't return an error, so it'd panic instead.

Here, I'm suggesting that we do a small breaking change by having
`rivermigrate.New` return a possible error. It's not great, but it's
something that'll easy for people to fix, and might help avoid some
future panics. An alternative possibility would be to add a new
`NewWithError` or something of that nature, but it'd make the API a
little uglier.

Along with that change, we also bring in a change to suggest migration
lines in case of an unknown one using Levenshtein distances. This is
aimed at preventing totally unactionable errors in the event of very
simple misspellings (e.g. "workflow" versus "workflows"). Cobra actually
already has a similar feature built-in for command misspellings.

I vendored in this repo [1] as a Levenshtein implementation and did a
little code clean up. I don't want to add another Go module dependency
for such a simple algorithm (it's just a dozen lines), and I believe
their MIT license should be compatible.

[1] https://github.com/agnivade/levenshtein/tree/master

CHANGELOG.md

@@ -12,6 +12,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - A new `river migrate-list` command is available which lists available migrations and which version a target database is migrated to. [PR #534](https://github.com/riverqueue/river/pull/534).
 - `river version` or `river --version` now prints River version information. [PR #537](https://github.com/riverqueue/river/pull/537).
 
+### Changed
+
+⚠️ Version 0.12.0 has a small breaking change in `rivermigrate`. As before, we try never to make breaking changes, but this one was deemed worth it because it's quite small and may help avoid panics.
+
+- **Breaking change:** `rivermigrate.New` now returns a possible error along with a migrator. An error may be returned, for example, when a migration line is configured that doesn't exist. [PR #XXX](https://github.com/riverqueue/river/pull/XXX).
+
+  ```go
+  # before
+	migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+
+  # after
+	migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+  if err != nil {
+      // handle error
+  }
+  ```
+
 ## [0.11.4] - 2024-08-20
 
 ### Fixed

cmd/river/rivercli/command.go

@@ -56,7 +56,7 @@ type CommandBase struct {
 	Out            io.Writer
 
 	GetBenchmarker func() BenchmarkerInterface
-	GetMigrator    func(config *rivermigrate.Config) MigratorInterface
+	GetMigrator    func(config *rivermigrate.Config) (MigratorInterface, error)
 }
 
 func (b *CommandBase) GetCommandBase() *CommandBase     { return b }
@@ -94,7 +94,7 @@ func RunCommand[TOpts CommandOpts](ctx context.Context, bundle *RunCommandBundle
 		// command doesn't take one.
 		case bundle.DatabaseURL == nil:
 			commandBase.GetBenchmarker = func() BenchmarkerInterface { panic("databaseURL was not set") }
-			commandBase.GetMigrator = func(config *rivermigrate.Config) MigratorInterface { panic("databaseURL was not set") }
+			commandBase.GetMigrator = func(config *rivermigrate.Config) (MigratorInterface, error) { panic("databaseURL was not set") }
 
 		case strings.HasPrefix(*bundle.DatabaseURL, uriScheme) ||
 			strings.HasPrefix(*bundle.DatabaseURL, uriSchemeAlias):
@@ -107,7 +107,7 @@ func RunCommand[TOpts CommandOpts](ctx context.Context, bundle *RunCommandBundle
 			driver := bundle.DriverProcurer.ProcurePgxV5(dbPool)
 
 			commandBase.GetBenchmarker = func() BenchmarkerInterface { return riverbench.NewBenchmarker(driver, commandBase.Logger) }
-			commandBase.GetMigrator = func(config *rivermigrate.Config) MigratorInterface { return rivermigrate.New(driver, config) }
+			commandBase.GetMigrator = func(config *rivermigrate.Config) (MigratorInterface, error) { return rivermigrate.New(driver, config) }
 
 		default:
 			return false, fmt.Errorf(

cmd/river/rivercli/river_cli.go

@@ -387,7 +387,12 @@ type migrateDown struct {
 }
 
 func (c *migrateDown) Run(ctx context.Context, opts *migrateOpts) (bool, error) {
-	res, err := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger}).Migrate(ctx, rivermigrate.DirectionDown, &rivermigrate.MigrateOpts{
+	migrator, err := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger})
+	if err != nil {
+		return false, err
+	}
+
+	res, err := migrator.Migrate(ctx, rivermigrate.DirectionDown, &rivermigrate.MigrateOpts{
 		DryRun:        opts.DryRun,
 		MaxSteps:      opts.MaxSteps,
 		TargetVersion: opts.TargetVersion,
@@ -470,7 +475,10 @@ func (c *migrateGet) Run(_ context.Context, opts *migrateGetOpts) (bool, error)
 	// other databases is added in the future. Unlike other migrate commands,
 	// this one doesn't take a `--database-url`, so we'd need a way of
 	// detecting the database type.
-	migrator := rivermigrate.New(c.DriverProcurer.ProcurePgxV5(nil), &rivermigrate.Config{Line: opts.Line, Logger: c.Logger})
+	migrator, err := rivermigrate.New(c.DriverProcurer.ProcurePgxV5(nil), &rivermigrate.Config{Line: opts.Line, Logger: c.Logger})
+	if err != nil {
+		return false, err
+	}
 
 	var migrations []rivermigrate.Migration
 	if opts.All {
@@ -534,7 +542,10 @@ type migrateList struct {
 }
 
 func (c *migrateList) Run(ctx context.Context, opts *migrateListOpts) (bool, error) {
-	migrator := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger})
+	migrator, err := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger})
+	if err != nil {
+		return false, err
+	}
 
 	allMigrations := migrator.AllVersions()
 
@@ -568,7 +579,12 @@ type migrateUp struct {
 }
 
 func (c *migrateUp) Run(ctx context.Context, opts *migrateOpts) (bool, error) {
-	res, err := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger}).Migrate(ctx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{
+	migrator, err := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger})
+	if err != nil {
+		return false, err
+	}
+
+	res, err := migrator.Migrate(ctx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{
 		DryRun:        opts.DryRun,
 		MaxSteps:      opts.MaxSteps,
 		TargetVersion: opts.TargetVersion,
@@ -600,7 +616,12 @@ type validate struct {
 }
 
 func (c *validate) Run(ctx context.Context, opts *validateOpts) (bool, error) {
-	res, err := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger}).Validate(ctx)
+	migrator, err := c.GetMigrator(&rivermigrate.Config{Line: opts.Line, Logger: c.Logger})
+	if err != nil {
+		return false, err
+	}
+
+	res, err := migrator.Validate(ctx)
 	if err != nil {
 		return false, err
 	}

cmd/river/rivercli/river_cli_test.go

@@ -177,7 +177,7 @@ func TestMigrateList(t *testing.T) {
 		migratorStub.allVersionsStub = func() []rivermigrate.Migration { return testMigrationAll }
 		migratorStub.existingVersionsStub = func(ctx context.Context) ([]rivermigrate.Migration, error) { return nil, nil }
 
-		cmd.GetCommandBase().GetMigrator = func(config *rivermigrate.Config) MigratorInterface { return migratorStub }
+		cmd.GetCommandBase().GetMigrator = func(config *rivermigrate.Config) (MigratorInterface, error) { return migratorStub, nil }
 
 		return cmd, &testBundle{
 			out:          out,
@@ -274,7 +274,7 @@ func withCommandBase[TCommand Command[TOpts], TOpts CommandOpts](t *testing.T, c
 		Logger: riversharedtest.Logger(t),
 		Out:    &out,
 
-		GetMigrator: func(config *rivermigrate.Config) MigratorInterface { return &MigratorStub{} },
+		GetMigrator: func(config *rivermigrate.Config) (MigratorInterface, error) { return &MigratorStub{}, nil },
 	})
 	return cmd, &out
 }

internal/cmd/testdbman/main.go

@@ -126,7 +126,11 @@ func createTestDatabases(ctx context.Context, out io.Writer) error {
 		}
 		defer dbPool.Close()
 
-		migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+		migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+		if err != nil {
+			return err
+		}
+
 		if _, err = migrator.Migrate(ctx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{}); err != nil {
 			return err
 		}

rivermigrate/example_migrate_database_sql_test.go

@@ -30,7 +30,10 @@ func Example_migrateDatabaseSQL() {
 	}
 	defer tx.Rollback()
 
-	migrator := rivermigrate.New(riverdatabasesql.New(dbPool), nil)
+	migrator, err := rivermigrate.New(riverdatabasesql.New(dbPool), nil)
+	if err != nil {
+		panic(err)
+	}
 
 	// Our test database starts with a full River schema. Drop it so that we can
 	// demonstrate working migrations. This isn't necessary outside this test.

rivermigrate/example_migrate_test.go

@@ -29,7 +29,10 @@ func Example_migrate() {
 	}
 	defer tx.Rollback(ctx)
 
-	migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+	migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+	if err != nil {
+		panic(err)
+	}
 
 	// Our test database starts with a full River schema. Drop it so that we can
 	// demonstrate working migrations. This isn't necessary outside this test.

rivermigrate/river_migrate.go

@@ -19,6 +19,7 @@ import (
 	"github.com/riverqueue/river/internal/util/dbutil"
 	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/rivershared/baseservice"
+	"github.com/riverqueue/river/rivershared/levenshtein"
 	"github.com/riverqueue/river/rivershared/util/maputil"
 	"github.com/riverqueue/river/rivershared/util/randutil"
 	"github.com/riverqueue/river/rivershared/util/sliceutil"
@@ -93,8 +94,11 @@ type Migrator[TTx any] struct {
 //	}
 //	defer dbPool.Close()
 //
-//	migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)
-func New[TTx any](driver riverdriver.Driver[TTx], config *Config) *Migrator[TTx] {
+//	migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+//	if err != nil {
+//		// handle error
+//	}
+func New[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Migrator[TTx], error) {
 	if config == nil {
 		config = &Config{}
 	}
@@ -115,7 +119,24 @@ func New[TTx any](driver riverdriver.Driver[TTx], config *Config) *Migrator[TTx]
 	}
 
 	if !slices.Contains(driver.GetMigrationLines(), line) {
-		panic("migration line does not exist: " + line)
+		const minLevenshteinDistance = 2
+
+		var suggestedLines []string
+		for _, existingLine := range driver.GetMigrationLines() {
+			if distance := levenshtein.ComputeDistance(existingLine, line); distance <= minLevenshteinDistance {
+				suggestedLines = append(suggestedLines, "`"+existingLine+"`")
+			}
+		}
+
+		errorStr := "migration line does not exist: " + line
+		switch {
+		case len(suggestedLines) == 1:
+			errorStr += fmt.Sprintf(" (did you mean %s?)", suggestedLines[0])
+		case len(suggestedLines) > 1:
+			errorStr += fmt.Sprintf(" (did you mean one of %v?)", strings.Join(suggestedLines, ", "))
+		}
+
+		return nil, errors.New(errorStr)
 	}
 
 	riverMigrations, err := migrationsFromFS(driver.GetMigrationFS(line), line)
@@ -129,7 +150,7 @@ func New[TTx any](driver riverdriver.Driver[TTx], config *Config) *Migrator[TTx]
 		driver:     driver,
 		line:       line,
 		migrations: validateAndInit(riverMigrations),
-	})
+	}), nil
 }
 
 // ExistingVersions gets the existing set of versions that have been migrated in

rivermigrate/river_migrate_test.go

@@ -48,12 +48,14 @@ func (d *driverWithAlternateLine) GetMigrationFS(line string) fs.FS {
 		return d.Driver.GetMigrationFS(line)
 	case migrationLineAlternate:
 		return migrationFS
+	case migrationLineAlternate + "2":
+		panic(line + " is only meant for testing line suggestions")
 	}
 	panic("migration line does not exist: " + line)
 }
 
 func (d *driverWithAlternateLine) GetMigrationLines() []string {
-	return append(d.Driver.GetMigrationLines(), migrationLineAlternate)
+	return append(d.Driver.GetMigrationLines(), migrationLineAlternate, migrationLineAlternate+"2")
 }
 
 func TestMigrator(t *testing.T) {
@@ -94,7 +96,8 @@ func TestMigrator(t *testing.T) {
 			tx:     tx,
 		}
 
-		migrator := New(bundle.driver, &Config{Logger: bundle.logger})
+		migrator, err := New(bundle.driver, &Config{Logger: bundle.logger})
+		require.NoError(t, err)
 		migrator.migrations = migrationsBundle.WithTestVersionsMap
 
 		return migrator, bundle
@@ -112,12 +115,41 @@ func TestMigrator(t *testing.T) {
 		t.Cleanup(func() { require.NoError(t, tx.Rollback()) })
 
 		driver := riverdatabasesql.New(stdPool)
-		migrator := New(driver, &Config{Logger: bundle.logger})
+		migrator, err := New(driver, &Config{Logger: bundle.logger})
+		require.NoError(t, err)
 		migrator.migrations = migrationsBundle.WithTestVersionsMap
 
 		return migrator, tx
 	}
 
+	t.Run("NewUnknownLine", func(t *testing.T) {
+		t.Parallel()
+
+		_, bundle := setup(t)
+
+		_, err := New(bundle.driver, &Config{Line: "unknown_line"})
+		require.EqualError(t, err, "migration line does not exist: unknown_line")
+
+		_, err = New(bundle.driver, &Config{Line: "mai"})
+		require.EqualError(t, err, "migration line does not exist: mai (did you mean `main`?)")
+
+		_, err = New(bundle.driver, &Config{Line: "maim"})
+		require.EqualError(t, err, "migration line does not exist: maim (did you mean `main`?)")
+
+		_, err = New(bundle.driver, &Config{Line: "maine"})
+		require.EqualError(t, err, "migration line does not exist: maine (did you mean `main`?)")
+
+		_, err = New(bundle.driver, &Config{Line: "ma"})
+		require.EqualError(t, err, "migration line does not exist: ma (did you mean `main`?)")
+
+		// Too far off.
+		_, err = New(bundle.driver, &Config{Line: "m"})
+		require.EqualError(t, err, "migration line does not exist: m")
+
+		_, err = New(bundle.driver, &Config{Line: "alternat"})
+		require.EqualError(t, err, "migration line does not exist: alternat (did you mean one of `alternate`, `alternate2`?)")
+	})
+
 	t.Run("AllVersions", func(t *testing.T) {
 		t.Parallel()
 
@@ -597,10 +629,11 @@ func TestMigrator(t *testing.T) {
 
 		// We have to reinitialize the alternateMigrator because the migrations bundle is
 		// set in the constructor.
-		alternateMigrator := New(bundle.driver, &Config{
+		alternateMigrator, err := New(bundle.driver, &Config{
 			Line:   migrationLineAlternate,
 			Logger: bundle.logger,
 		})
+		require.NoError(t, err)
 
 		res, err := alternateMigrator.MigrateTx(ctx, bundle.tx, DirectionUp, &MigrateOpts{})
 		require.NoError(t, err)
@@ -633,10 +666,11 @@ func TestMigrator(t *testing.T) {
 		_, err := migrator.MigrateTx(ctx, bundle.tx, DirectionDown, &MigrateOpts{TargetVersion: 4})
 		require.NoError(t, err)
 
-		alternateMigrator := New(bundle.driver, &Config{
+		alternateMigrator, err := New(bundle.driver, &Config{
 			Line:   migrationLineAlternate,
 			Logger: bundle.logger,
 		})
+		require.NoError(t, err)
 
 		// Alternate line not allowed because `river_job.line` doesn't exist.
 		_, err = alternateMigrator.MigrateTx(ctx, bundle.tx, DirectionUp, &MigrateOpts{})

rivershared/levenshtein/License.txt

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2015 Agniva De Sarker
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.