cmd: migrate command

cmd/migrate.go

@@ -0,0 +1,107 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"go/types"
+	"strconv"
+
+	migrate "github.com/rubenv/sql-migrate"
+	"github.com/spf13/cobra"
+	"github.com/stellar/go/support/config"
+	"github.com/stellar/go/support/log"
+	"github.com/stellar/wallet-backend/internal/db"
+)
+
+type migrateCmd struct{}
+
+func (c *migrateCmd) Command() *cobra.Command {
+	var databaseURL string
+	cfgOpts := config.ConfigOptions{
+		{
+			Name:        "database-url",
+			Usage:       "Database connection URL",
+			OptType:     types.String,
+			ConfigKey:   &databaseURL,
+			FlagDefault: "postgres://postgres@localhost:5432/wallet-backend?sslmode=disable",
+			Required:    true,
+		},
+	}
+
+	migrateCmd := &cobra.Command{
+		Use:   "migrate",
+		Short: "Schema migration helpers",
+		PersistentPreRun: func(_ *cobra.Command, _ []string) {
+			cfgOpts.Require()
+			if err := cfgOpts.SetValues(); err != nil {
+				log.Fatalf("Error setting values of config options: %s", err.Error())
+			}
+		},
+	}
+
+	migrateUpCmd := cobra.Command{
+		Use:   "up",
+		Short: "Migrates database up [count] migrations",
+		Args:  cobra.MaximumNArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			var count int
+			if len(args) > 0 {
+				var err error
+				count, err = strconv.Atoi(args[0])
+				if err != nil {
+					log.Fatalf("Invalid [count] argument: %s", args[0])
+				}
+			}
+
+			if err := executeMigrations(cmd.Context(), databaseURL, migrate.Up, count); err != nil {
+				log.Fatalf("Error executing migrate up: %v", err)
+			}
+		},
+	}
+
+	migrateDownCmd := &cobra.Command{
+		Use:   "down [count]",
+		Short: "Migrates database down [count] migrations",
+		Args:  cobra.ExactArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			count, err := strconv.Atoi(args[0])
+			if err != nil {
+				log.Fatalf("Invalid [count] argument: %s", args[0])
+			}
+
+			if err := executeMigrations(cmd.Context(), databaseURL, migrate.Down, count); err != nil {
+				log.Fatalf("Error executing migrate down: %v", err)
+			}
+		},
+	}
+
+	migrateCmd.AddCommand(&migrateUpCmd)
+	migrateCmd.AddCommand(migrateDownCmd)
+
+	if err := cfgOpts.Init(migrateCmd); err != nil {
+		log.Fatalf("Error initializing a config option: %s", err.Error())
+	}
+
+	return migrateCmd
+}
+
+func executeMigrations(ctx context.Context, databaseURL string, direction migrate.MigrationDirection, count int) error {
+	numMigrationsRun, err := db.Migrate(databaseURL, direction, count)
+	if err != nil {
+		return fmt.Errorf("migrating database: %w", err)
+	}
+
+	if numMigrationsRun == 0 {
+		log.Ctx(ctx).Info("No migrations applied.")
+	} else {
+		log.Ctx(ctx).Infof("Successfully applied %d migrations %s.", numMigrationsRun, migrationDirectionStr(direction))
+	}
+	return nil
+}
+
+func migrationDirectionStr(direction migrate.MigrationDirection) string {
+	if direction == migrate.Up {
+		return "up"
+	}
+	return "down"
+}

cmd/root.go

@@ -31,4 +31,5 @@ func init() {
 
 	rootCmd.AddCommand((&serveCmd{}).Command())
 	rootCmd.AddCommand((&ingestCmd{}).Command())
+	rootCmd.AddCommand((&migrateCmd{}).Command())
 }

internal/db/dbtest/dbtest.go

@@ -24,3 +24,8 @@ func Open(t *testing.T) *dbtest.DB {
 
 	return db
 }
+
+func OpenWithoutMigrations(t *testing.T) *dbtest.DB {
+	db := dbtest.Postgres(t)
+	return db
+}

internal/db/migrate.go

@@ -0,0 +1,26 @@
+package db
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	migrate "github.com/rubenv/sql-migrate"
+	"github.com/stellar/wallet-backend/internal/db/migrations"
+)
+
+func Migrate(databaseURL string, direction migrate.MigrationDirection, count int) (int, error) {
+	dbConnectionPool, err := OpenDBConnectionPool(databaseURL)
+	if err != nil {
+		return 0, fmt.Errorf("connecting to the database: %w", err)
+	}
+	defer dbConnectionPool.Close()
+
+	m := migrate.HttpFileSystemMigrationSource{FileSystem: http.FS(migrations.FS)}
+	ctx := context.Background()
+	db, err := dbConnectionPool.SqlDB(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("fetching sql.DB: %w", err)
+	}
+	return migrate.ExecMax(db, dbConnectionPool.DriverName(), m, direction, count)
+}

internal/db/migrate_test.go

@@ -0,0 +1,108 @@
+package db
+
+import (
+	"context"
+	"io/fs"
+	"testing"
+
+	migrate "github.com/rubenv/sql-migrate"
+	"github.com/stellar/wallet-backend/internal/db/dbtest"
+	"github.com/stellar/wallet-backend/internal/db/migrations"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMigrate_up_1(t *testing.T) {
+	dbt := dbtest.OpenWithoutMigrations(t)
+	defer dbt.Close()
+
+	dbConnectionPool, err := OpenDBConnectionPool(dbt.DSN)
+	require.NoError(t, err)
+	defer dbConnectionPool.Close()
+
+	ctx := context.Background()
+
+	n, err := Migrate(dbt.DSN, migrate.Up, 1)
+	require.NoError(t, err)
+	assert.Equal(t, 1, n)
+
+	ids := []string{}
+	err = dbConnectionPool.SelectContext(ctx, &ids, "SELECT id FROM gorp_migrations")
+	require.NoError(t, err)
+	wantIDs := []string{"2024-04-29.0-initial.sql"}
+	assert.Equal(t, wantIDs, ids)
+}
+
+func TestMigrate_up_2_down_1(t *testing.T) {
+	dbt := dbtest.OpenWithoutMigrations(t)
+	defer dbt.Close()
+
+	dbConnectionPool, err := OpenDBConnectionPool(dbt.DSN)
+	require.NoError(t, err)
+	defer dbConnectionPool.Close()
+
+	ctx := context.Background()
+
+	n, err := Migrate(dbt.DSN, migrate.Up, 2)
+	require.NoError(t, err)
+	assert.Equal(t, 2, n)
+
+	ids := []string{}
+	err = dbConnectionPool.SelectContext(ctx, &ids, "SELECT id FROM gorp_migrations")
+	require.NoError(t, err)
+	wantIDs := []string{
+		"2024-04-29.0-initial.sql",
+		"2024-04-29.1-accounts.sql",
+	}
+	assert.Equal(t, wantIDs, ids)
+
+	n, err = Migrate(dbt.DSN, migrate.Down, 1)
+	require.NoError(t, err)
+	assert.Equal(t, 1, n)
+
+	err = dbConnectionPool.SelectContext(ctx, &ids, "SELECT id FROM gorp_migrations")
+	require.NoError(t, err)
+	wantIDs = []string{
+		"2024-04-29.0-initial.sql",
+	}
+	assert.Equal(t, wantIDs, ids)
+}
+
+func TestMigrate_upall_down_all(t *testing.T) {
+	db := dbtest.OpenWithoutMigrations(t)
+	defer db.Close()
+
+	dbConnectionPool, err := OpenDBConnectionPool(db.DSN)
+	require.NoError(t, err)
+	defer dbConnectionPool.Close()
+
+	// Get number of files in the migrations directory
+	var count int
+	err = fs.WalkDir(migrations.FS, ".", func(path string, d fs.DirEntry, err error) error {
+		require.NoError(t, err)
+		if !d.IsDir() {
+			count++
+		}
+		return nil
+	})
+	require.NoError(t, err)
+
+	n, err := Migrate(db.DSN, migrate.Up, 0)
+	require.NoError(t, err)
+	require.Equal(t, count, n)
+
+	ctx := context.Background()
+
+	var migrationsCount int
+	err = dbConnectionPool.GetContext(ctx, &migrationsCount, "SELECT COUNT(*) FROM gorp_migrations")
+	require.NoError(t, err)
+	assert.Equal(t, count, migrationsCount)
+
+	n, err = Migrate(db.DSN, migrate.Down, count)
+	require.NoError(t, err)
+	require.Equal(t, count, n)
+
+	err = dbConnectionPool.GetContext(ctx, &migrationsCount, "SELECT COUNT(*) FROM gorp_migrations")
+	require.NoError(t, err)
+	assert.Equal(t, 0, migrationsCount)
+}