pawsql/internal/postgres/provisioner.go

// Package postgres provisions PawSQL-managed PostgreSQL containers through the Docker CLI.
package postgres

import (
	"bytes"
	"context"
	"fmt"
	"log/slog"
	"net"
	"os"
	"os/exec"
	"strings"
	"time"

	config "cloud.campbellwireless.net/git/barkstack/barkfile-parser"
)

const managedDatabaseLabel = "io.barkstack.pawsql.database"

// Provisioner ensures configured PostgreSQL containers exist and exposes their
// loopback-published PostgreSQL address to PawSQL's static router.
type Provisioner struct {
	DockerPath string
	Logger     *slog.Logger
}

// NewProvisioner creates a Docker CLI-backed provisioner.
func NewProvisioner(logger *slog.Logger) *Provisioner {
	if logger == nil {
		logger = slog.Default()
	}
	return &Provisioner{DockerPath: "docker", Logger: logger}
}

// Ensure provisions each PostgreSQL-backed database and returns configuration
// with its discovered upstream addresses. Existing containers and volumes are
// adopted only when they carry PawSQL's matching ownership label.
func (p *Provisioner) Ensure(ctx context.Context, cfg config.Config) (config.Config, error) {
	resolved := cfg
	resolved.Databases = append([]config.DatabaseConfig(nil), cfg.Databases...)
	for index := range resolved.Databases {
		database := &resolved.Databases[index]
		if database.Postgres == nil {
			continue
		}
		address, err := p.EnsureDatabase(ctx, database.Name, *database.Postgres)
		if err != nil {
			return config.Config{}, fmt.Errorf("provision database %q: %w", database.Name, err)
		}
		database.Upstream = address
	}
	return resolved, nil
}

// EnsureDatabase creates or starts one managed PostgreSQL database and waits
// until its loopback-published address accepts connections.
func (p *Provisioner) EnsureDatabase(ctx context.Context, database string, postgres config.PostgresConfig) (string, error) {
	name := containerName(database)
	if _, err := p.run(ctx, nil, "volume", "create", postgres.Volume); err != nil {
		return "", fmt.Errorf("create volume %q: %w", postgres.Volume, err)
	}

	label, exists, err := p.containerLabel(ctx, name)
	if err != nil {
		return "", err
	}
	if !exists {
		password, ok := os.LookupEnv(postgres.PasswordEnv)
		if !ok || password == "" {
			return "", fmt.Errorf("environment variable %q is required to create the container", postgres.PasswordEnv)
		}
		p.Logger.Info("creating PostgreSQL container", "database", database, "container", name, "image", postgres.Image, "volume", postgres.Volume)
		mount := postgresDataMount(postgres.Image)
		_, err := p.run(ctx, []string{"POSTGRES_PASSWORD=" + password},
			"container", "create",
			"--name", name,
			"--label", managedDatabaseLabel+"="+database,
			"--env", "POSTGRES_DB="+database,
			"--env", "POSTGRES_USER="+database,
			"--env", "POSTGRES_PASSWORD",
			"--volume", postgres.Volume+":"+mount,
			"--publish", "127.0.0.1::5432",
			postgres.Image,
		)
		if err != nil {
			return "", fmt.Errorf("create container %q: %w", name, err)
		}
	} else if label != database {
		return "", fmt.Errorf("container %q belongs to %q, not PawSQL database %q", name, label, database)
	}

	running, err := p.containerRunning(ctx, name)
	if err != nil {
		return "", err
	}
	if !running {
		p.Logger.Info("starting PostgreSQL container", "database", database, "container", name)
		if _, err := p.run(ctx, nil, "container", "start", name); err != nil {
			return "", fmt.Errorf("start container %q: %w", name, err)
		}
	}
	port, err := p.hostPort(ctx, name)
	if err != nil {
		return "", err
	}
	address := net.JoinHostPort("127.0.0.1", port)
	if err := waitForPostgres(ctx, address); err != nil {
		return "", fmt.Errorf("wait for PostgreSQL container %q: %w", name, err)
	}
	return address, nil
}

// StopDatabase stops a managed PostgreSQL container without removing its data volume.
func (p *Provisioner) StopDatabase(ctx context.Context, database string) error {
	name := containerName(database)
	label, exists, err := p.containerLabel(ctx, name)
	if err != nil {
		return err
	}
	if !exists {
		return nil
	}
	if label != database {
		return fmt.Errorf("container %q belongs to %q, not PawSQL database %q", name, label, database)
	}
	running, err := p.containerRunning(ctx, name)
	if err != nil {
		return err
	}
	if !running {
		return nil
	}
	p.Logger.Info("stopping idle PostgreSQL container", "database", database, "container", name)
	if _, err := p.run(ctx, nil, "container", "stop", name); err != nil {
		return fmt.Errorf("stop container %q: %w", name, err)
	}
	return nil
}

func waitForPostgres(ctx context.Context, address string) error {
	deadline := time.NewTimer(30 * time.Second)
	defer deadline.Stop()
	for {
		connection, err := (&net.Dialer{Timeout: time.Second}).DialContext(ctx, "tcp", address)
		if err == nil {
			return connection.Close()
		}
		retry := time.NewTimer(200 * time.Millisecond)
		select {
		case <-ctx.Done():
			retry.Stop()
			return ctx.Err()
		case <-deadline.C:
			retry.Stop()
			return err
		case <-retry.C:
		}
	}
}

func (p *Provisioner) containerLabel(ctx context.Context, name string) (label string, exists bool, err error) {
	output, err := p.run(ctx, nil, "container", "inspect", "--format", "{{ index .Config.Labels \""+managedDatabaseLabel+"\" }}", name)
	if err != nil {
		if strings.Contains(err.Error(), "No such container") {
			return "", false, nil
		}
		return "", false, fmt.Errorf("inspect container %q: %w", name, err)
	}
	return strings.TrimSpace(output), true, nil
}

func (p *Provisioner) containerRunning(ctx context.Context, name string) (bool, error) {
	output, err := p.run(ctx, nil, "container", "inspect", "--format", "{{.State.Running}}", name)
	if err != nil {
		return false, fmt.Errorf("inspect container state %q: %w", name, err)
	}
	return strings.TrimSpace(output) == "true", nil
}

func (p *Provisioner) hostPort(ctx context.Context, name string) (string, error) {
	output, err := p.run(ctx, nil, "container", "inspect", "--format", "{{(index (index .NetworkSettings.Ports \"5432/tcp\") 0).HostPort}}", name)
	if err != nil {
		return "", fmt.Errorf("inspect PostgreSQL port %q: %w", name, err)
	}
	port := strings.TrimSpace(output)
	if port == "" {
		return "", fmt.Errorf("container %q has no published PostgreSQL port", name)
	}
	return port, nil
}

func (p *Provisioner) run(ctx context.Context, environment []string, args ...string) (string, error) {
	path := p.DockerPath
	if path == "" {
		path = "docker"
	}
	command := exec.CommandContext(ctx, path, args...)
	command.Env = append(os.Environ(), environment...)
	var stdout, stderr bytes.Buffer
	command.Stdout = &stdout
	command.Stderr = &stderr
	if err := command.Run(); err != nil {
		return "", fmt.Errorf("docker %s: %w: %s", strings.Join(args, " "), err, strings.TrimSpace(stderr.String()))
	}
	return stdout.String(), nil
}

func containerName(database string) string {
	return "pawsql-" + database
}

func postgresDataMount(image string) string {
	if image == "postgres:18" {
		return "/var/lib/postgresql"
	}
	return "/var/lib/postgresql/data"
}