WIP.

cmd/migrate/main.go

@@ -5,6 +5,7 @@ import (
 	"net/url"
 	"os"
 
+	"github.com/fatih/color"
 	"github.com/go-nacelle/config/v3"
 	"github.com/go-nacelle/log/v2"
 	"github.com/go-nacelle/pgutil"
@@ -26,6 +27,7 @@ var (
 	migrationDirectory string
 	databaseURL        string
 	defaultDatabaseURL = pgutil.BuildDatabaseURL()
+	noColor            bool
 	logger             log.Logger
 )
 
@@ -52,6 +54,20 @@ func init() {
 		"",
 		fmt.Sprintf("The database connection URL (default %s)", masked),
 	)
+
+	rootCmd.PersistentFlags().BoolVarP(
+		&noColor,
+		"no-color",
+		"",
+		false,
+		"Disable color output",
+	)
+
+	rootCmd.PersistentPreRun = func(cmd *cobra.Command, args []string) {
+		if noColor {
+			color.NoColor = true
+		}
+	}
 }
 
 func initLogger() (err error) {

cmd/migrate/state.go

@@ -3,7 +3,10 @@ package main
 import (
 	"context"
 	"fmt"
+	"strings"
 
+	"github.com/fatih/color"
+	"github.com/go-nacelle/pgutil"
 	"github.com/spf13/cobra"
 )
 
@@ -23,18 +26,93 @@ func state(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
+	definitions := runner.Definitions()
+
 	logs, err := runner.MigrationLogs(context.Background())
 	if err != nil {
 		return err
 	}
+	logMap := map[int]pgutil.MigrationLog{}
+	for _, log := range logs {
+		logMap[log.MigrationID] = log
+	}
 
-	if len(logs) == 0 {
-		fmt.Printf("Empty database.\n")
-	} else {
-		for _, log := range logs {
-			fmt.Printf("> %v %v %v\n", log.MigrationID, log.Reverse, log.Success)
+	maxDefinitionLen := 0
+	for _, definition := range definitions {
+		if len(definition.Name) > maxDefinitionLen {
+			maxDefinitionLen = len(definition.Name)
+		}
+	}
+
+	type migrationError struct {
+		definition   pgutil.Definition
+		errorMessage string
+	}
+	errorMessages := []migrationError{}
+
+	for _, definition := range definitions {
+		log, exists := logMap[definition.ID]
+		color, statusEmoji, statusText := definitionStatus(log, exists)
+
+		color.Printf(
+			"%s %04d: %s\t%s\n",
+			statusEmoji,
+			definition.ID,
+			definition.Name+strings.Repeat(" ", maxDefinitionLen-len(definition.Name)),
+			statusText,
+		)
+
+		if exists && log.ErrorMessage != nil {
+			errorMessages = append(errorMessages, migrationError{
+				definition:   definition,
+				errorMessage: *log.ErrorMessage,
+			})
+		}
+	}
+
+	if len(errorMessages) > 0 {
+		fmt.Println("\nErrors:")
+
+		for _, message := range errorMessages {
+			fmt.Printf("  %04d: %s\n", message.definition.ID, message.errorMessage)
 		}
 	}
 
 	return nil
 }
+
+const (
+	emojiApplied    = "✅"
+	emojiError      = "❌"
+	emojiUnknown    = "❓"
+	emojiReverse    = "↩️"
+	emojiNotApplied = "  "
+)
+
+func definitionStatus(log pgutil.MigrationLog, exists bool) (_ *color.Color, statusEmoji string, statusText string) {
+	if !exists {
+		return color.New(color.FgCyan), emojiNotApplied, "Not applied"
+	}
+
+	if log.Success != nil {
+		if *log.Success {
+			if !log.Reverse {
+				return color.New(color.FgGreen), emojiApplied, "Successfully applied"
+			} else {
+				return color.New(color.FgYellow), emojiReverse, "Successfully un-apply"
+			}
+		} else {
+			if !log.Reverse {
+				return color.New(color.FgRed), emojiError, "Failed most recent apply"
+			} else {
+				return color.New(color.FgRed), emojiError, "Failed most recent un-apply"
+			}
+		}
+	}
+
+	if !log.Reverse {
+		return color.New(color.FgMagenta), emojiUnknown, "Attempted apply (unknown status)"
+	} else {
+		return color.New(color.FgMagenta), emojiUnknown, "Attempted un-apply (unknown status)"
+	}
+}

cmd/migrate/undo.go

@@ -31,4 +31,4 @@ func undo(cmd *cobra.Command, args []string) error {
 	}
 
 	return runner.Undo(context.Background(), migrationID)
-}
+}

cmd/migrate/up.go

@@ -2,13 +2,16 @@ package main
 
 import (
 	"context"
+	"fmt"
+	"strconv"
 
 	"github.com/spf13/cobra"
 )
 
 var upCmd = &cobra.Command{
-	Use:   "up",
-	Short: "Run all schema migrations",
+	Use:   "up [migration_id]?",
+	Short: "Run migrations up to and including the specified migration ID",
+	Args:  cobra.MaximumNArgs(1),
 	RunE:  up,
 }
 
@@ -22,5 +25,14 @@ func up(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
-	return runner.ApplyAll(context.Background())
+	if len(args) == 0 {
+		return runner.ApplyAll(context.Background())
+	}
+
+	migrationID, err := strconv.Atoi(args[0])
+	if err != nil {
+		return fmt.Errorf("invalid migration ID: %v", err)
+	}
+
+	return runner.Apply(context.Background(), migrationID)
 }

go.mod

@@ -20,7 +20,7 @@ require (
 	github.com/BurntSushi/toml v1.4.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/derision-test/glock v1.1.0 // indirect
-	github.com/fatih/color v1.15.0 // indirect
+	github.com/fatih/color v1.17.0 // indirect
 	github.com/fatih/structtag v1.2.0 // indirect
 	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/go-nacelle/process/v2 v2.1.0 // indirect

go.sum

@@ -25,6 +25,8 @@ github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25Kn
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fatih/color v1.15.0 h1:kOqh6YHBtK8aywxGerMG2Eq3H6Qgoqeo13Bk2Mv/nBs=
 github.com/fatih/color v1.15.0/go.mod h1:0h5ZqXfHYED7Bhv2ZJamyIOUej9KtShiJESRwBDUSsw=
+github.com/fatih/color v1.17.0 h1:GlRw1BRJxkpqUCBKzKOw098ed57fEsKeNjpTe3cSjK4=
+github.com/fatih/color v1.17.0/go.mod h1:YZ7TlrGPkiz6ku9fK3TLD/pl3CpsiFyu8N92HLgmosI=
 github.com/fatih/structtag v1.2.0 h1:/OdNE99OxoI/PqaW/SuSK9uxxT3f/tcSZgon/ssNSx4=
 github.com/fatih/structtag v1.2.0/go.mod h1:mBJUNpUnHmRKrKlQQlmCrh5PuhftFbNv8Ys4/aAZl94=
 github.com/frankban/quicktest v1.14.3/go.mod h1:mgiwOwqx65TmIk1wJ6Q7wvnVMocbUorkibMOrVTHZps=

migration_reader_embed.go

@@ -3,5 +3,5 @@ package pgutil
 import "embed"
 
 func NewEmbedMigrationReader(fs embed.FS) MigrationReader {
-	return newFilesystemMigrationReader(fs)
+	return newFilesystemMigrationReader("<embed>", fs)
 }

migration_reader_filesystem.go

@@ -1,6 +1,7 @@
 package pgutil
 
 import (
+	"fmt"
 	"io"
 	"io/fs"
 	"net/http"
@@ -12,22 +13,28 @@ import (
 )
 
 type FilesystemMigrationReader struct {
-	fs fs.FS
+	name string
+	fs   fs.FS
 }
 
 func NewFilesystemMigrationReader(dirname string) MigrationReader {
-	return newFilesystemMigrationReader(os.DirFS(dirname))
+	return newFilesystemMigrationReader(dirname, os.DirFS(dirname))
 }
 
-func newFilesystemMigrationReader(fs fs.FS) MigrationReader {
+func newFilesystemMigrationReader(name string, fs fs.FS) MigrationReader {
 	return &FilesystemMigrationReader{
-		fs: fs,
+		name: name,
+		fs:   fs,
 	}
 }
 
 func (r *FilesystemMigrationReader) ReadAll() (definitions []RawDefinition, _ error) {
 	root, err := http.FS(r.fs).Open("/")
 	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, fmt.Errorf("migration directory %q does not exist", r.name)
+		}
+
 		return nil, err
 	}
 	defer root.Close()

migration_runner.go

@@ -37,6 +37,10 @@ func NewMigrationRunner(db DB, reader MigrationReader, logger nacelle.Logger) (*
 	}, nil
 }
 
+func (r *Runner) Definitions() []Definition {
+	return r.definitions
+}
+
 func (r *Runner) ApplyAll(ctx context.Context) error {
 	return r.apply(ctx, r.definitions)
 }
@@ -145,20 +149,24 @@ func (r *Runner) applyDefinitions(ctx context.Context, definitions []Definition,
 			}
 
 			if err := r.withMigrationLog(ctx, definition, reverse, func(_ int) error {
-				return r.db.WithTransaction(ctx, func(tx DB) error {
-					query, direction := definition.UpQuery, "up"
-					if reverse {
-						query, direction = definition.DownQuery, "down"
-					}
-
-					r.logger.InfoWithFields(log.LogFields{
-						"id":        definition.ID,
-						"name":      definition.Name,
-						"direction": direction,
-					}, "Applying migration")
-
-					return tx.Exec(ctx, query)
+				query, direction := definition.UpQuery, "up"
+				if reverse {
+					query, direction = definition.DownQuery, "down"
+				}
+
+				logger := r.logger.WithFields(log.LogFields{
+					"id":        definition.ID,
+					"name":      definition.Name,
+					"direction": direction,
 				})
+				logger.Info("Applying migration")
+
+				if err := r.db.WithTransaction(ctx, func(tx DB) error { return tx.Exec(ctx, query) }); err != nil {
+					logger.ErrorWithFields(log.LogFields{"error": err}, "Failed to apply migration")
+					return err
+				}
+
+				return nil
 			}); err != nil {
 				return err
 			}
@@ -341,18 +349,19 @@ func (r *Runner) createIndexConcurrently(ctx context.Context, definition Definit
 //
 //
 
-type migrationLog struct {
-	MigrationID int
-	Reverse     bool
-	Success     *bool
+type MigrationLog struct {
+	MigrationID  int
+	Reverse      bool
+	Success      *bool
+	ErrorMessage *string
 }
 
-var scanMigrationLogs = NewSliceScanner(func(s Scanner) (ms migrationLog, _ error) {
-	err := s.Scan(&ms.MigrationID, &ms.Reverse, &ms.Success)
+var scanMigrationLogs = NewSliceScanner(func(s Scanner) (ms MigrationLog, _ error) {
+	err := s.Scan(&ms.MigrationID, &ms.Reverse, &ms.Success, &ms.ErrorMessage)
 	return ms, err
 })
 
-func (r *Runner) MigrationLogs(ctx context.Context) (map[int]migrationLog, error) {
+func (r *Runner) MigrationLogs(ctx context.Context) (map[int]MigrationLog, error) {
 	if err := r.ensureMigrationLogsTable(ctx); err != nil {
 		return nil, err
 	}
@@ -367,7 +376,8 @@ func (r *Runner) MigrationLogs(ctx context.Context) (map[int]migrationLog, error
 		SELECT
 			migration_id,
 			reverse,
-			success
+			success,
+			error_message
 		FROM ranked_migration_logs
 		WHERE rank = 1
 		ORDER BY migration_id
@@ -376,7 +386,7 @@ func (r *Runner) MigrationLogs(ctx context.Context) (map[int]migrationLog, error
 		return nil, err
 	}
 
-	logMap := map[int]migrationLog{}
+	logMap := map[int]MigrationLog{}
 	for _, state := range migrationLogs {
 		logMap[state.MigrationID] = state
 	}