feat(03-01): add shared postgres pool and plugin migrations
Open one pgx stdlib *sql.DB, hand it to GORM, and run per-plugin gormigrate sets with isolated history tables after an ICU pl-PL check. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
86
lagoon/commands.go
Normal file
86
lagoon/commands.go
Normal file
@@ -0,0 +1,86 @@
|
||||
package lagoon
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"git.golem15.com/golem15/summercms/backpack"
|
||||
"git.golem15.com/golem15/summercms/bonfire"
|
||||
"git.golem15.com/golem15/summercms/party"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// RuntimeCommands returns migrate, migrate:rollback and migrate:status.
|
||||
// Serve is registered once the HTTP layer exists.
|
||||
func RuntimeCommands(app *backpack.App, plugins []party.Plugin) []bonfire.Command {
|
||||
return []bonfire.Command{
|
||||
{
|
||||
Name: "migrate",
|
||||
Description: "Run plugin migrations in dependency order",
|
||||
Run: func(ctx context.Context, in bonfire.Input, out bonfire.Output) error {
|
||||
return withDB(ctx, app, func(gdb *gorm.DB) error {
|
||||
if err := Migrate(gdb, plugins); err != nil {
|
||||
return err
|
||||
}
|
||||
out.Success("migrations applied")
|
||||
return nil
|
||||
})
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "migrate:rollback",
|
||||
Description: "Roll back the last migration of a plugin",
|
||||
Flags: []bonfire.Flag{{
|
||||
Name: "plugin",
|
||||
Description: "Plugin ID whose last migration to roll back",
|
||||
}},
|
||||
Run: func(ctx context.Context, in bonfire.Input, out bonfire.Output) error {
|
||||
plugin, _ := in.Flag("plugin")
|
||||
return withDB(ctx, app, func(gdb *gorm.DB) error {
|
||||
if err := RollbackLast(gdb, plugins, plugin); err != nil {
|
||||
return err
|
||||
}
|
||||
if plugin == "" {
|
||||
plugin = lastMigrationPlugin(plugins)
|
||||
}
|
||||
out.Success(fmt.Sprintf("rolled back last migration of %s", plugin))
|
||||
return nil
|
||||
})
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "migrate:status",
|
||||
Description: "Show per-plugin migration history",
|
||||
Run: func(ctx context.Context, in bonfire.Input, out bonfire.Output) error {
|
||||
return withDB(ctx, app, func(gdb *gorm.DB) error {
|
||||
rows, err := Status(gdb, plugins)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tableRows := make([][]string, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
ids := "(none)"
|
||||
if len(row.IDs) > 0 {
|
||||
ids = fmt.Sprintf("%d: %s", len(row.IDs), row.IDs[len(row.IDs)-1])
|
||||
}
|
||||
tableRows = append(tableRows, []string{row.Plugin, row.Table, ids})
|
||||
}
|
||||
out.Table([]string{"plugin", "table", "applied"}, tableRows)
|
||||
return nil
|
||||
})
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func withDB(ctx context.Context, app *backpack.App, fn func(*gorm.DB) error) error {
|
||||
sqlDB, gdb, err := OpenFromApp(ctx, app)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer sqlDB.Close()
|
||||
if err := Publish(app, sqlDB, gdb); err != nil {
|
||||
return err
|
||||
}
|
||||
return fn(gdb)
|
||||
}
|
||||
136
lagoon/connection.go
Normal file
136
lagoon/connection.go
Normal file
@@ -0,0 +1,136 @@
|
||||
package lagoon
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"git.golem15.com/golem15/summercms/backpack"
|
||||
"git.golem15.com/golem15/summercms/compass"
|
||||
_ "github.com/jackc/pgx/v5/stdlib"
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
requiredLocaleProvider = "i"
|
||||
requiredICULocale = "pl-PL"
|
||||
)
|
||||
|
||||
// Open pings dsn through pgx stdlib, requires Postgres 16 ICU pl-PL, and
|
||||
// returns that exact *sql.DB plus a GORM handle opened on it.
|
||||
//
|
||||
// Phase 11 owns a separate pgxpool.Pool for River LISTEN/NOTIFY. Do not
|
||||
// create that listener pool here; application queries share this *sql.DB.
|
||||
func Open(ctx context.Context, dsn string) (*sql.DB, *gorm.DB, error) {
|
||||
dsn = strings.TrimSpace(dsn)
|
||||
if dsn == "" {
|
||||
return nil, nil, fmt.Errorf("lagoon: database.dsn is empty (set SUMMER_DATABASE__DSN)")
|
||||
}
|
||||
sqlDB, err := sql.Open("pgx", dsn)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("lagoon: open postgres: %w", err)
|
||||
}
|
||||
if err := sqlDB.PingContext(ctx); err != nil {
|
||||
_ = sqlDB.Close()
|
||||
return nil, nil, fmt.Errorf("lagoon: ping postgres: %w", err)
|
||||
}
|
||||
if err := CheckLocale(ctx, sqlDB); err != nil {
|
||||
_ = sqlDB.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
gdb, err := gormFromSQL(sqlDB)
|
||||
if err != nil {
|
||||
_ = sqlDB.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return sqlDB, gdb, nil
|
||||
}
|
||||
|
||||
// Use pings an existing pool, requires ICU pl-PL, and returns a GORM handle
|
||||
// opened on that exact *sql.DB. Callers that already hold a pool (tests,
|
||||
// the app boot seam) must not open a second connection.
|
||||
func Use(ctx context.Context, sqlDB *sql.DB) (*gorm.DB, error) {
|
||||
if sqlDB == nil {
|
||||
return nil, fmt.Errorf("lagoon: sql db is nil")
|
||||
}
|
||||
if err := sqlDB.PingContext(ctx); err != nil {
|
||||
return nil, fmt.Errorf("lagoon: ping postgres: %w", err)
|
||||
}
|
||||
if err := CheckLocale(ctx, sqlDB); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return gormFromSQL(sqlDB)
|
||||
}
|
||||
|
||||
func gormFromSQL(sqlDB *sql.DB) (*gorm.DB, error) {
|
||||
gdb, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("lagoon: gorm open: %w", err)
|
||||
}
|
||||
got, err := gdb.DB()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("lagoon: gorm sql handle: %w", err)
|
||||
}
|
||||
if got != sqlDB {
|
||||
return nil, fmt.Errorf("lagoon: GORM is not using the shared *sql.DB")
|
||||
}
|
||||
return gdb, nil
|
||||
}
|
||||
|
||||
// OpenFromApp reads database.dsn from app config and opens the shared pool.
|
||||
func OpenFromApp(ctx context.Context, app *backpack.App) (*sql.DB, *gorm.DB, error) {
|
||||
if app == nil || app.Config == nil {
|
||||
return nil, nil, fmt.Errorf("lagoon: app config is missing")
|
||||
}
|
||||
return Open(ctx, DSN(app.Config))
|
||||
}
|
||||
|
||||
// DSN returns database.dsn from layered config (env SUMMER_DATABASE__DSN).
|
||||
func DSN(cfg *compass.Config) string {
|
||||
if cfg == nil {
|
||||
return ""
|
||||
}
|
||||
return strings.TrimSpace(cfg.String("database.dsn"))
|
||||
}
|
||||
|
||||
// Publish stores the shared SQL pool and GORM handle on the app. Both
|
||||
// handles refer to the same *sql.DB.
|
||||
func Publish(app *backpack.App, sqlDB *sql.DB, gdb *gorm.DB) error {
|
||||
if app == nil {
|
||||
return fmt.Errorf("lagoon: app is nil")
|
||||
}
|
||||
if sqlDB == nil || gdb == nil {
|
||||
return fmt.Errorf("lagoon: database handles are nil")
|
||||
}
|
||||
if err := app.Publish(sqlDB); err != nil {
|
||||
return err
|
||||
}
|
||||
return app.Publish(gdb)
|
||||
}
|
||||
|
||||
// CheckLocale fails unless the connected database uses ICU locale pl-PL.
|
||||
func CheckLocale(ctx context.Context, db *sql.DB) error {
|
||||
if db == nil {
|
||||
return fmt.Errorf("lagoon: sql db is nil")
|
||||
}
|
||||
var provider, icu string
|
||||
err := db.QueryRowContext(ctx, `
|
||||
SELECT datlocprovider::text, COALESCE(daticulocale, '')
|
||||
FROM pg_database
|
||||
WHERE datname = current_database()`).Scan(&provider, &icu)
|
||||
if err != nil {
|
||||
return fmt.Errorf("lagoon: read database locale: %w", err)
|
||||
}
|
||||
return checkLocale(provider, icu)
|
||||
}
|
||||
|
||||
func checkLocale(provider, icu string) error {
|
||||
provider = strings.TrimSpace(provider)
|
||||
icu = strings.TrimSpace(icu)
|
||||
if provider == requiredLocaleProvider && icu == requiredICULocale {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("lagoon: database locale must be ICU pl-PL (datlocprovider=%q, daticulocale=%q); got provider %q locale %q. Create the database with: CREATE DATABASE ... TEMPLATE template0 ENCODING 'UTF8' LOCALE_PROVIDER icu ICU_LOCALE 'pl-PL'", requiredLocaleProvider, requiredICULocale, provider, icu)
|
||||
}
|
||||
155
lagoon/migrations.go
Normal file
155
lagoon/migrations.go
Normal file
@@ -0,0 +1,155 @@
|
||||
package lagoon
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"unicode"
|
||||
|
||||
"git.golem15.com/golem15/summercms/pact"
|
||||
"git.golem15.com/golem15/summercms/party"
|
||||
"github.com/go-gormigrate/gormigrate/v2"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const historyTablePrefix = "summer_migrations_"
|
||||
|
||||
// HistoryTableName returns the isolated gormigrate table for pluginID.
|
||||
func HistoryTableName(pluginID string) (string, error) {
|
||||
if err := validatePluginID(pluginID); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return historyTablePrefix + strings.ReplaceAll(pluginID, ".", "_"), nil
|
||||
}
|
||||
|
||||
func validatePluginID(id string) error {
|
||||
if id == "" {
|
||||
return fmt.Errorf("lagoon: plugin id is empty")
|
||||
}
|
||||
for _, r := range id {
|
||||
if r == '.' || unicode.IsLower(r) || unicode.IsDigit(r) {
|
||||
continue
|
||||
}
|
||||
return fmt.Errorf("lagoon: plugin id %q is not a valid history table name", id)
|
||||
}
|
||||
if strings.Contains(id, "..") || strings.HasPrefix(id, ".") || strings.HasSuffix(id, ".") {
|
||||
return fmt.Errorf("lagoon: plugin id %q is not a valid history table name", id)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func migrator(gdb *gorm.DB, pluginID string, migrations []*gormigrate.Migration) (*gormigrate.Gormigrate, error) {
|
||||
table, err := HistoryTableName(pluginID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return gormigrate.New(gdb, &gormigrate.Options{
|
||||
TableName: table,
|
||||
UseTransaction: true,
|
||||
}, migrations), nil
|
||||
}
|
||||
|
||||
// Migrate runs each plugin's HasMigrations set in party.Activate order.
|
||||
func Migrate(gdb *gorm.DB, plugins []party.Plugin) error {
|
||||
if gdb == nil {
|
||||
return fmt.Errorf("lagoon: gorm db is nil")
|
||||
}
|
||||
for _, p := range plugins {
|
||||
hm, ok := p.(pact.HasMigrations)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
migrations := hm.Migrations()
|
||||
if len(migrations) == 0 {
|
||||
continue
|
||||
}
|
||||
m, err := migrator(gdb, p.ID(), migrations)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := m.Migrate(); err != nil {
|
||||
return fmt.Errorf("lagoon: migrate %s: %w", p.ID(), err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RollbackLast rolls back the last migration of pluginID.
|
||||
func RollbackLast(gdb *gorm.DB, plugins []party.Plugin, pluginID string) error {
|
||||
if gdb == nil {
|
||||
return fmt.Errorf("lagoon: gorm db is nil")
|
||||
}
|
||||
pluginID = strings.TrimSpace(pluginID)
|
||||
if pluginID == "" {
|
||||
pluginID = lastMigrationPlugin(plugins)
|
||||
}
|
||||
if pluginID == "" {
|
||||
return fmt.Errorf("lagoon: no plugin migrations to roll back")
|
||||
}
|
||||
for _, p := range plugins {
|
||||
if p.ID() != pluginID {
|
||||
continue
|
||||
}
|
||||
hm, ok := p.(pact.HasMigrations)
|
||||
if !ok {
|
||||
return fmt.Errorf("lagoon: plugin %q has no migrations", pluginID)
|
||||
}
|
||||
m, err := migrator(gdb, p.ID(), hm.Migrations())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := m.RollbackLast(); err != nil {
|
||||
return fmt.Errorf("lagoon: rollback %s: %w", pluginID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("lagoon: plugin %q is not activated", pluginID)
|
||||
}
|
||||
|
||||
func lastMigrationPlugin(plugins []party.Plugin) string {
|
||||
for i := len(plugins) - 1; i >= 0; i-- {
|
||||
p := plugins[i]
|
||||
hm, ok := p.(pact.HasMigrations)
|
||||
if !ok || len(hm.Migrations()) == 0 {
|
||||
continue
|
||||
}
|
||||
return p.ID()
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// StatusRow is one plugin's recorded migration ids.
|
||||
type StatusRow struct {
|
||||
Plugin string
|
||||
Table string
|
||||
IDs []string
|
||||
}
|
||||
|
||||
// Status reads each plugin history table.
|
||||
func Status(gdb *gorm.DB, plugins []party.Plugin) ([]StatusRow, error) {
|
||||
if gdb == nil {
|
||||
return nil, fmt.Errorf("lagoon: gorm db is nil")
|
||||
}
|
||||
var rows []StatusRow
|
||||
for _, p := range plugins {
|
||||
if _, ok := p.(pact.HasMigrations); !ok {
|
||||
continue
|
||||
}
|
||||
table, err := HistoryTableName(p.ID())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row := StatusRow{Plugin: p.ID(), Table: table}
|
||||
if !gdb.Migrator().HasTable(table) {
|
||||
rows = append(rows, row)
|
||||
continue
|
||||
}
|
||||
if err := gdb.Table(table).Order("id").Pluck("id", &row.IDs).Error; err != nil {
|
||||
return nil, fmt.Errorf("lagoon: status %s: %w", p.ID(), err)
|
||||
}
|
||||
if row.IDs == nil {
|
||||
row.IDs = []string{}
|
||||
}
|
||||
rows = append(rows, row)
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
97
lagoon/migrations_test.go
Normal file
97
lagoon/migrations_test.go
Normal file
@@ -0,0 +1,97 @@
|
||||
package lagoon
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.golem15.com/golem15/summercms/bonfire"
|
||||
)
|
||||
|
||||
func TestHistoryTableName(t *testing.T) {
|
||||
got, err := HistoryTableName("golem15.user")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != "summer_migrations_golem15_user" {
|
||||
t.Fatalf("got %q", got)
|
||||
}
|
||||
got, err = HistoryTableName("golem15.fonoteka")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != "summer_migrations_golem15_fonoteka" {
|
||||
t.Fatalf("got %q", got)
|
||||
}
|
||||
if _, err := HistoryTableName("golem15.user;drop"); err == nil {
|
||||
t.Fatal("want invalid plugin id error")
|
||||
}
|
||||
if _, err := HistoryTableName(""); err == nil {
|
||||
t.Fatal("want empty plugin id error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckLocaleMessage(t *testing.T) {
|
||||
if err := checkLocale("i", "pl-PL"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err := checkLocale("c", "en_US")
|
||||
if err == nil {
|
||||
t.Fatal("want locale error")
|
||||
}
|
||||
msg := err.Error()
|
||||
for _, want := range []string{"ICU", "pl-PL", "CREATE DATABASE", "LOCALE_PROVIDER icu", "en_US"} {
|
||||
if !strings.Contains(msg, want) {
|
||||
t.Fatalf("missing %q in %s", want, msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRuntimeCommandsRegisterBareAndColonNames(t *testing.T) {
|
||||
cmds := RuntimeCommands(nil, nil)
|
||||
names := map[string]bool{}
|
||||
for _, c := range cmds {
|
||||
names[c.Name] = true
|
||||
}
|
||||
for _, want := range []string{"migrate", "migrate:rollback", "migrate:status"} {
|
||||
if !names[want] {
|
||||
t.Fatalf("missing %s", want)
|
||||
}
|
||||
}
|
||||
root, err := bonfire.NewRoot("app", cmds, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
root.SetArgs([]string{"migrate", "--help"})
|
||||
if err := root.Execute(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoAutoMigrate(t *testing.T) {
|
||||
entries, err := os.ReadDir(".")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, entry := range entries {
|
||||
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".go") || strings.HasSuffix(entry.Name(), "_test.go") {
|
||||
continue
|
||||
}
|
||||
body, err := os.ReadFile(entry.Name())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if bytes.Contains(body, []byte("AutoMigrate")) {
|
||||
t.Fatalf("%s must not call AutoMigrate", entry.Name())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenRequiresDSN(t *testing.T) {
|
||||
_, _, err := Open(t.Context(), " ")
|
||||
if err == nil || !strings.Contains(err.Error(), "database.dsn") {
|
||||
t.Fatalf("got %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user