Add pgroll migrate subcommand (#465)

Add a `pgroll migrate` subcommand.

## Documentation
`pgroll migrate` applies all outstanding migrations from a source
directory to the target database.

Assuming that migrations up to and including migration
`40_create_enum_type` from the [example migrations
directory](https://github.com/xataio/pgroll/tree/main/examples) have
been applied, running:

```
$ pgroll migrate examples/
```

will apply migrations from `41_add_enum_column` onwards to the target
database.

If the `--complete` flag is passed to `pgroll migrate` the final
migration to be applied will be completed. Otherwise the final migration
will be left active (started but not completed).

## Notes:
* If no migrations have yet been applied to the target database,
`migrate` applies all of the migrations in the source directory.
* This PR removes the `pgroll bootstrap` command
(#414) as it is equivalent to
running `pgroll migrate <directory> --complete` against a fresh
database.

Part of #446

Author: andrew-farries

Makefile

@@ -31,7 +31,7 @@ ledger:
 examples: ledger
 	@go build
 	@./pgroll init
-	@./pgroll bootstrap examples
+	@./pgroll migrate examples --complete
 	@go clean
 
 test:

cmd/bootstrap.go

@@ -0,0 +1,77 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package cmd
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/spf13/cobra"
+)
+
+func migrateCmd() *cobra.Command {
+	var complete bool
+
+	migrateCmd := &cobra.Command{
+		Use:     "migrate <directory>",
+		Short:   "Apply outstanding migrations from a directory to a database",
+		Example: "migrate ./migrations",
+		Args:    cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := cmd.Context()
+			migrationsDir := args[0]
+
+			m, err := NewRoll(ctx)
+			if err != nil {
+				return err
+			}
+			defer m.Close()
+
+			latestVersion, err := m.State().LatestVersion(ctx, m.Schema())
+			if err != nil {
+				return fmt.Errorf("unable to determine latest version: %w", err)
+			}
+
+			active, err := m.State().IsActiveMigrationPeriod(ctx, m.Schema())
+			if err != nil {
+				return fmt.Errorf("unable to determine active migration period: %w", err)
+			}
+			if active {
+				return fmt.Errorf("migration %q is active and must be completed first", *latestVersion)
+			}
+
+			info, err := os.Stat(migrationsDir)
+			if err != nil {
+				return fmt.Errorf("failed to stat directory: %w", err)
+			}
+			if !info.IsDir() {
+				return fmt.Errorf("migrations directory %q is not a directory", migrationsDir)
+			}
+
+			migs, err := m.UnappliedMigrations(ctx, os.DirFS(migrationsDir))
+			if err != nil {
+				return fmt.Errorf("failed to get migrations to apply: %w", err)
+			}
+
+			if len(migs) == 0 {
+				fmt.Println("database is up to date; no migrations to apply")
+				return nil
+			}
+
+			// Run all migrations after the latest version up to the final migration,
+			// completing each one.
+			for _, mig := range migs[:len(migs)-1] {
+				if err := runMigration(ctx, m, mig, true); err != nil {
+					return fmt.Errorf("failed to run migration file %q: %w", mig.Name, err)
+				}
+			}
+
+			// Run the final migration, completing it only if requested.
+			return runMigration(ctx, m, migs[len(migs)-1], complete)
+		},
+	}
+
+	migrateCmd.Flags().BoolVarP(&complete, "complete", "c", false, "complete the final migration rather than leaving it active")
+
+	return migrateCmd
+}

cmd/migrate.go

@@ -76,7 +76,7 @@ func Execute() error {
 	rootCmd.AddCommand(analyzeCmd)
 	rootCmd.AddCommand(initCmd)
 	rootCmd.AddCommand(statusCmd)
-	rootCmd.AddCommand(bootstrapCmd)
+	rootCmd.AddCommand(migrateCmd())
 	rootCmd.AddCommand(pullCmd())
 
 	return rootCmd.Execute()

cmd/root.go

@@ -6,8 +6,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"path/filepath"
-	"strings"
 
 	"github.com/pterm/pterm"
 	"github.com/spf13/cobra"
@@ -52,12 +50,16 @@ func runMigrationFromFile(ctx context.Context, m *roll.Roll, fileName string, co
 		return err
 	}
 
+	return runMigration(ctx, m, migration, complete)
+}
+
+func runMigration(ctx context.Context, m *roll.Roll, migration *migrations.Migration, complete bool) error {
 	sp, _ := pterm.DefaultSpinner.WithText("Starting migration...").Start()
 	cb := func(n int64) {
 		sp.UpdateText(fmt.Sprintf("%d records complete...", n))
 	}
 
-	err = m.Start(ctx, migration, cb)
+	err := m.Start(ctx, migration, cb)
 	if err != nil {
 		sp.Fail(fmt.Sprintf("Failed to start migration: %s", err))
 		return err
@@ -71,9 +73,6 @@ func runMigrationFromFile(ctx context.Context, m *roll.Roll, fileName string, co
 	}
 
 	version := migration.Name
-	if version == "" {
-		version = strings.TrimSuffix(filepath.Base(fileName), filepath.Ext(fileName))
-	}
 	viewName := roll.VersionedSchemaName(flags.Schema(), version)
 	msg := fmt.Sprintf("New version of the schema available under the postgres %q schema", viewName)
 	sp.Success(msg)

cmd/start.go

@@ -623,6 +623,21 @@ Migrations cannot be rolled back once completed. Attempting to roll back a migra
 
 :warning: Before running `pgroll rollback` ensure that any new versions of applications that depend on the new database schema are no longer live. Prematurely running `pgroll rollback` can cause downtime of new application instances that depend on the new schema.
 
+
+### Migrate
+
+`pgroll migrate` applies all outstanding migrations from a directory to the target database.
+
+Assuming that migrations up to and including migration `40_create_enum_type` from the [example migrations](https://github.com/xataio/pgroll/tree/main/examples) directory have been applied, running:
+
+```
+$ pgroll migrate examples/
+```
+
+will apply migrations from `41_add_enum_column` onwards to the target database.
+
+If the `--complete` flag is passed to `pgroll migrate` the final migration to be applied will be completed. Otherwise the final migration will be left active (started but not completed).
+
 ### Status
 
 `pgroll status` shows the current status of `pgroll` within a given schema:

docs/README.md

@@ -24,6 +24,8 @@ const (
 	DefaultBackfillDelay     time.Duration = 0
 )
 
+var ErrMismatchedMigration = fmt.Errorf("remote migration does not match local migration")
+
 type Roll struct {
 	pgConn db.DB
 
@@ -144,6 +146,11 @@ func (m *Roll) PgConn() db.DB {
 	return m.pgConn
 }
 
+// State returns the state instance the Roll instance is acting on
+func (m *Roll) State() *state.State {
+	return m.state
+}
+
 // Schema returns the schema the Roll instance is acting on
 func (m *Roll) Schema() string {
 	return m.schema

pkg/roll/roll.go

@@ -0,0 +1,81 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package roll
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+
+	"github.com/xataio/pgroll/pkg/migrations"
+)
+
+// UnappliedMigrations returns a slice of unapplied migrations from `dir`,
+// lexicographically ordered by filename. Applying each of the returned
+// migrations in order will bring the database up to date with `dir`.
+//
+// If the local order of migrations does not match the order of migrations in
+// the schema history, an `ErrMismatchedMigration` error is returned.
+func (m *Roll) UnappliedMigrations(ctx context.Context, dir fs.FS) ([]*migrations.Migration, error) {
+	latestVersion, err := m.State().LatestVersion(ctx, m.Schema())
+	if err != nil {
+		return nil, fmt.Errorf("determining latest version: %w", err)
+	}
+
+	files, err := fs.Glob(dir, "*.json")
+	if err != nil {
+		return nil, fmt.Errorf("reading directory: %w", err)
+	}
+
+	history, err := m.State().SchemaHistory(ctx, m.Schema())
+	if err != nil {
+		return nil, fmt.Errorf("reading schema history: %w", err)
+	}
+
+	// Find the index of the first unapplied migration
+	var idx int
+	if latestVersion != nil {
+		for _, file := range files {
+			migration, err := openAndReadMigrationFile(dir, file)
+			if err != nil {
+				return nil, fmt.Errorf("reading migration file %q: %w", file, err)
+			}
+
+			remoteMigration := history[idx].Migration
+			if remoteMigration.Name != migration.Name {
+				return nil, fmt.Errorf("%w: remote=%q, local=%q", ErrMismatchedMigration, remoteMigration.Name, migration.Name)
+			}
+
+			idx++
+			if migration.Name == *latestVersion {
+				break
+			}
+		}
+	}
+
+	// Return all unapplied migrations
+	migs := make([]*migrations.Migration, 0, len(files))
+	for _, file := range files[idx:] {
+		migration, err := openAndReadMigrationFile(dir, file)
+		if err != nil {
+			return nil, fmt.Errorf("reading migration file %q: %w", file, err)
+		}
+		migs = append(migs, migration)
+	}
+
+	return migs, nil
+}
+
+func openAndReadMigrationFile(dir fs.FS, filename string) (*migrations.Migration, error) {
+	file, err := dir.Open(filename)
+	if err != nil {
+		return nil, err
+	}
+
+	migration, err := migrations.ReadMigration(file)
+	if err != nil {
+		return nil, err
+	}
+
+	return migration, nil
+}