feat: add postgres setup and migration to ftl serve (#518)

Fixes #499 

When docker isn't running
```bash
ftl serve

info: Checking for FTL database
error: Cannot connect to the Docker daemon at unix:///Users/wesbillman/.orbstack/run/docker.sock. Is the docker daemon running?
: exit status 1
ftl: error: exit status 1
```

When the database hasn't been created yet
```bash
ftl serve

info: Checking for FTL database
info: Creating FTL database
info: 7f1fc8b7e07c983efdba2ef7e207ac0d7e82eb704bcd7e990ff5573839dafa8e
info: Waiting for ftl-db to be healthy
info: Initializing FTL schema
info: Applying: 001_init.sql
info: Starting 1 controller(s) and 10 runner(s)
```

When the database already exists
```bash
ftl serve
info: Checking for FTL database
info: Starting 1 controller(s) and 10 runner(s)
```

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

cmd/ftl/cmd_serve.go

@@ -8,11 +8,14 @@ import (
 	"os"
 	"path/filepath"
 	"strconv"
+	"strings"
+	"time"
 
 	"github.com/alecthomas/errors"
 	"github.com/alecthomas/kong"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/TBD54566975/ftl/backend/common/exec"
 	"github.com/TBD54566975/ftl/backend/common/log"
 	"github.com/TBD54566975/ftl/backend/controller"
 	"github.com/TBD54566975/ftl/backend/runner"
@@ -24,8 +27,16 @@ type serveCmd struct {
 	Runners     int      `short:"r" help:"Number of runners to start." default:"10"`
 }
 
+const ftlContainerName = "ftl-db"
+
 func (s *serveCmd) Run(ctx context.Context) error {
 	logger := log.FromContext(ctx)
+
+	dsn, err := setupDB(ctx)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
 	logger.Infof("Starting %d controller(s) and %d runner(s)", s.Controllers, s.Runners)
 
 	wg, ctx := errgroup.WithContext(ctx)
@@ -38,6 +49,7 @@ func (s *serveCmd) Run(ctx context.Context) error {
 		controllerAddresses = append(controllerAddresses, nextBind)
 		config := controller.Config{
 			Bind: nextBind,
+			DSN:  dsn,
 		}
 		if err := kong.ApplyDefaults(&config); err != nil {
 			return errors.WithStack(err)
@@ -105,6 +117,71 @@ func (s *serveCmd) Run(ctx context.Context) error {
 	return nil
 }
 
+func setupDB(ctx context.Context) (string, error) {
+	logger := log.FromContext(ctx)
+	logger.Infof("Checking for FTL database")
+
+	nameFlag := fmt.Sprintf("name=^/%s$", ftlContainerName)
+	output, err := exec.Capture(ctx, ".", "docker", "ps", "-a", "--filter", nameFlag, "--format", "{{.Names}}")
+	if err != nil {
+		logger.Errorf(err, "%s", output)
+		return "", errors.WithStack(err)
+	}
+
+	recreate := false
+
+	if len(output) == 0 {
+		logger.Infof("Creating FTL database")
+
+		err = exec.Command(ctx, logger.GetLevel(), "./", "docker", "run",
+			"-d", // run detached so we can follow with other commands
+			"--name", ftlContainerName,
+			"--user", "postgres",
+			"--restart", "always",
+			"-e", "POSTGRES_PASSWORD=secret",
+			"-p", "5432", // dynamically allocate port
+			"--health-cmd=pg_isready",
+			"--health-interval=1s",
+			"--health-timeout=60s",
+			"--health-retries=60",
+			"--health-start-period=80s",
+			"postgres:latest", "postgres",
+		).Run()
+
+		if err != nil {
+			return "", errors.WithStack(err)
+		}
+
+		err = pollContainerHealth(ctx, ftlContainerName, 10*time.Second)
+		if err != nil {
+			return "", err
+		}
+
+		recreate = true
+	}
+
+	// grab the port from docker for this container
+	port, err := exec.Capture(ctx, ".", "docker", "inspect", "--format", "{{ (index (index .NetworkSettings.Ports \"5432/tcp\") 0).HostPort }}", ftlContainerName)
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+
+	dsn := fmt.Sprintf("postgres://postgres:secret@localhost:%s/%s?sslmode=disable", strings.TrimSpace(string(port)), ftlContainerName)
+	dsnFlag := fmt.Sprintf("--dsn=%s", dsn)
+
+	if recreate {
+		logger.Infof("Initializing FTL schema")
+		err = exec.Command(ctx, logger.GetLevel(), ".", "ftl-initdb", "--recreate", dsnFlag).Run()
+	} else {
+		err = exec.Command(ctx, logger.GetLevel(), ".", "ftl-initdb", dsnFlag).Run()
+	}
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+
+	return dsn, nil
+}
+
 func incrementPort(baseURL *url.URL) (*url.URL, error) {
 	newURL := *baseURL
 
@@ -116,3 +193,28 @@ func incrementPort(baseURL *url.URL) (*url.URL, error) {
 	newURL.Host = fmt.Sprintf("%s:%d", baseURL.Hostname(), newPort+1)
 	return &newURL, nil
 }
+
+func pollContainerHealth(ctx context.Context, containerName string, timeout time.Duration) error {
+	logger := log.FromContext(ctx)
+	logger.Infof("Waiting for %s to be healthy", containerName)
+
+	pollCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	for {
+		select {
+		case <-pollCtx.Done():
+			return errors.New("timed out waiting for container to be healthy")
+		case <-time.After(1 * time.Millisecond):
+			output, err := exec.Capture(pollCtx, ".", "docker", "inspect", "--format", "{{.State.Health.Status}}", containerName)
+			if err != nil {
+				return errors.WithStack(err)
+			}
+
+			status := strings.TrimSpace(string(output))
+			if status == "healthy" {
+				return nil
+			}
+		}
+	}
+}