Scaffold tomswallapi control-plane service

internal/database/database.go

@@ -0,0 +1,90 @@
+// Package database provides the Postgres connection pool and schema migrations
+// for tomswallapi.
+package database
+
+import (
+	"context"
+	"embed"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+//go:embed migrations/*.sql
+var migrationsFS embed.FS
+
+// DB wraps a pgx connection pool.
+type DB struct {
+	Pool *pgxpool.Pool
+}
+
+// New opens a connection pool and verifies connectivity.
+func New(ctx context.Context, dsn string) (*DB, error) {
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		return nil, fmt.Errorf("creating pool: %w", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("pinging database: %w", err)
+	}
+	return &DB{Pool: pool}, nil
+}
+
+// Close releases the pool.
+func (db *DB) Close() { db.Pool.Close() }
+
+// Migrate applies any pending embedded SQL migrations in lexical order. Each
+// migration file is recorded in schema_migrations and applied at most once.
+func (db *DB) Migrate(ctx context.Context) error {
+	if _, err := db.Pool.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS schema_migrations (
+			version    TEXT PRIMARY KEY,
+			applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		)`); err != nil {
+		return fmt.Errorf("creating schema_migrations: %w", err)
+	}
+
+	entries, err := migrationsFS.ReadDir("migrations")
+	if err != nil {
+		return fmt.Errorf("reading migrations: %w", err)
+	}
+	var files []string
+	for _, e := range entries {
+		if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
+			files = append(files, e.Name())
+		}
+	}
+	sort.Strings(files)
+
+	for _, name := range files {
+		var exists bool
+		if err := db.Pool.QueryRow(ctx,
+			`SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, name,
+		).Scan(&exists); err != nil {
+			return fmt.Errorf("checking migration %s: %w", name, err)
+		}
+		if exists {
+			continue
+		}
+
+		body, err := migrationsFS.ReadFile("migrations/" + name)
+		if err != nil {
+			return fmt.Errorf("reading migration %s: %w", name, err)
+		}
+
+		if err := pgx.BeginFunc(ctx, db.Pool, func(tx pgx.Tx) error {
+			if _, err := tx.Exec(ctx, string(body)); err != nil {
+				return fmt.Errorf("applying %s: %w", name, err)
+			}
+			_, err := tx.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, name)
+			return err
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}