feat: ftl migrate command (#2033)

Fixes #1839

- Migration files are embedded in the binary, so the binary version
matches the schema.
- Uses `DATABASE_URL` (same as dbmate) or `--dsn` to specify the
connection.

<img width="909" alt="image"
src="https://github.com/TBD54566975/ftl/assets/31338/81b6fc6c-ede4-42c4-a5d5-1ed2d7b02342">

backend/controller/sql/database_integration_test.go

@@ -3,6 +3,7 @@
 package sql_test
 
 import (
+	"fmt"
 	"testing"
 
 	in "github.com/TBD54566975/ftl/integration"
@@ -23,3 +24,19 @@ func TestDatabase(t *testing.T) {
 		in.QueryRow("testdb", "SELECT data FROM requests", "hello"),
 	)
 }
+
+func TestMigrate(t *testing.T) {
+	dbName := "ftl_test"
+	dbUri := fmt.Sprintf("postgres://postgres:secret@localhost:15432/%s?sslmode=disable", dbName)
+
+	q := func() in.Action {
+		return in.QueryRow(dbName, "SELECT version FROM schema_migrations WHERE version = '20240704103403'", "20240704103403")
+	}
+
+	in.RunWithoutController(t, "",
+		in.DropDBAction(t, dbName),
+		in.Fail(q(), "Should fail because the database does not exist."),
+		in.Exec("ftl", "migrate", "--dsn", dbUri),
+		q(),
+	)
+}

backend/controller/sql/databasetesting/devel.go

@@ -62,7 +62,7 @@ func CreateForDevel(ctx context.Context, dsn string, recreate bool) (*pgxpool.Po
 
 	_, _ = conn.Exec(ctx, fmt.Sprintf("CREATE DATABASE %q", config.Database)) //nolint:errcheck // PG doesn't support "IF NOT EXISTS" so instead we just ignore any error.
 
-	err = sql.Migrate(ctx, dsn)
+	err = sql.Migrate(ctx, dsn, log.Debug)
 	if err != nil {
 		return nil, err
 	}

backend/controller/sql/migrate.go

@@ -18,7 +18,7 @@ import (
 var migrationSchema embed.FS
 
 // Migrate the database.
-func Migrate(ctx context.Context, dsn string) error {
+func Migrate(ctx context.Context, dsn string, logLevel log.Level) error {
 	u, err := url.Parse(dsn)
 	if err != nil {
 		return fmt.Errorf("invalid DSN: %w", err)
@@ -31,7 +31,7 @@ func Migrate(ctx context.Context, dsn string) error {
 
 	db := dbmate.New(u)
 	db.FS = migrationSchema
-	db.Log = log.FromContext(ctx).Scope("migrate").WriterAt(log.Debug)
+	db.Log = log.FromContext(ctx).Scope("migrate").WriterAt(logLevel)
 	db.MigrationsDir = []string{"schema"}
 	err = db.CreateAndMigrate()
 	if err != nil {

cmd/ftl/cmd_migrate.go

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/TBD54566975/ftl/backend/controller/sql"
+	"github.com/TBD54566975/ftl/internal/log"
+)
+
+type migrateCmd struct {
+	DSN string `help:"DSN for the database." default:"postgres://localhost:15432/ftl?sslmode=disable&user=postgres&password=secret" env:"DATABASE_URL"`
+}
+
+func (c *migrateCmd) Run(ctx context.Context) error {
+	logger := log.FromContext(ctx)
+	logger.Infof("Migrating database")
+	err := sql.Migrate(ctx, c.DSN, log.Info)
+	if err != nil {
+		return fmt.Errorf("failed to migrate database: %w", err)
+	}
+	return nil
+}

cmd/ftl/main.go

@@ -47,6 +47,7 @@ type CLI struct {
 	Box      boxCmd      `cmd:"" help:"Build a self-contained Docker container for running a set of module."`
 	BoxRun   boxRunCmd   `cmd:"" hidden:"" help:"Run FTL inside an ftl-in-a-box container"`
 	Deploy   deployCmd   `cmd:"" help:"Build and deploy all modules found in the specified directories."`
+	Migrate  migrateCmd  `cmd:"" help:"Run a database migration, if required, based on the migration table."`
 	Download downloadCmd `cmd:"" help:"Download a deployment."`
 	Secret   secretCmd   `cmd:"" help:"Manage secrets."`
 	Config   configCmd   `cmd:"" help:"Manage configuration."`

integration/actions.go

@@ -345,6 +345,17 @@ func CreateDBAction(module, dbName string, isTest bool) Action {
 	}
 }
 
+func terminateDanglingConnections(t testing.TB, db *sql.DB, dbName string) {
+	t.Helper()
+
+	_, err := db.Exec(`
+		SELECT pid, pg_terminate_backend(pid)
+		FROM pg_stat_activity
+		WHERE datname = $1 AND pid <> pg_backend_pid()`,
+		dbName)
+	assert.NoError(t, err)
+}
+
 func CreateDB(t testing.TB, module, dbName string, isTestDb bool) {
 	// insert test suffix if needed when actually setting up db
 	if isTestDb {
@@ -368,18 +379,34 @@ func CreateDB(t testing.TB, module, dbName string, isTestDb bool) {
 	assert.NoError(t, err, "failed to create database")
 
 	t.Cleanup(func() {
-		// Terminate any dangling connections.
-		_, err := db.Exec(`
-				SELECT pid, pg_terminate_backend(pid)
-				FROM pg_stat_activity
-				WHERE datname = $1 AND pid <> pg_backend_pid()`,
-			dbName)
-		assert.NoError(t, err)
+		terminateDanglingConnections(t, db, dbName)
 		_, err = db.Exec("DROP DATABASE " + dbName)
 		assert.NoError(t, err)
 	})
 }
 
+func DropDBAction(t testing.TB, dbName string) Action {
+	return func(t testing.TB, ic TestContext) {
+		DropDB(t, dbName)
+	}
+}
+
+func DropDB(t testing.TB, dbName string) {
+	Infof("Dropping database %s", dbName)
+	db, err := sql.Open("pgx", "postgres://postgres:secret@localhost:15432/postgres?sslmode=disable")
+	assert.NoError(t, err, "failed to open database connection")
+
+	terminateDanglingConnections(t, db, dbName)
+
+	_, err = db.Exec("DROP DATABASE IF EXISTS " + dbName)
+	assert.NoError(t, err, "failed to delete existing database")
+
+	t.Cleanup(func() {
+		err := db.Close()
+		assert.NoError(t, err)
+	})
+}
+
 // Create a directory in the working directory
 func Mkdir(dir string) Action {
 	return func(t testing.TB, ic TestContext) {