Files
summercms/modules/lagoon/migrations.go
Jakub Zych 0bc5c77097 feat(11-01): add the conga job framework on River with transactional dispatch
- Pin River v0.47.0 (riverdatabasesql, rivertype) and tidy the example modules
- lagoon.Migrate runs the summercms.conga set: River schema v7 and summer_jobs
  with the apparatus columns plus an internal river_job_id link
- conga.Manager.Dispatch writes the summer_jobs row (status IN_PROGRESS) and the
  River job on the caller's *sql.Tx; a rollback leaves neither
- conga.Job wraps typed job functions so plugins never import River
- conga.StartWorker runs one client on riverdatabasesql.NewWithPgxListener with
  a single-connection LISTEN pool; the final failed attempt sets ERROR
- TestListenPickupLatency: 30s poll, pickup under 1s; poll-only control 2s miss
2026-09-29 15:02:04 +02:00

196 lines
5.0 KiB
Go

package lagoon
import (
"errors"
"fmt"
"strings"
"unicode"
"git.golem15.com/golem15/summercms/modules/lagoon/attach"
"git.golem15.com/golem15/summercms/modules/pact"
"git.golem15.com/golem15/summercms/modules/party"
"github.com/go-gormigrate/gormigrate/v2"
"gorm.io/gorm"
)
const historyTablePrefix = "summer_migrations_"
var (
// ErrUnknownPlugin is returned when migrate:rollback names a plugin that is not activated.
ErrUnknownPlugin = errors.New("lagoon: plugin is not activated")
// ErrNoMigrations is returned when there is no plugin migration set to roll back.
ErrNoMigrations = errors.New("lagoon: no plugin migrations to roll back")
)
// 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 the framework-owned system_files, backend-admin and job-queue
// (QueueMigrations) sets first, then 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")
}
m, err := migrator(gdb, "summercms.attach", attach.Migrations)
if err != nil {
return err
}
if err := m.Migrate(); err != nil {
return fmt.Errorf("lagoon: migrate system_files: %w", err)
}
admin, err := migrator(gdb, "summercms.cabana", BackendAdminMigrations)
if err != nil {
return err
}
if err := admin.Migrate(); err != nil {
return fmt.Errorf("lagoon: migrate backend admin: %w", err)
}
sqlDB, err := gdb.DB()
if err != nil {
return fmt.Errorf("lagoon: migrate queue: %w", err)
}
queue, err := migrator(gdb, QueueHistoryID, QueueMigrations(sqlDB))
if err != nil {
return err
}
if err := queue.Migrate(); err != nil {
return fmt.Errorf("lagoon: migrate queue: %w", err)
}
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 ErrNoMigrations
}
var target party.Plugin
for _, p := range plugins {
if p.ID() == pluginID {
target = p
break
}
}
if target == nil {
return fmt.Errorf("%w: %q", ErrUnknownPlugin, pluginID)
}
hm, ok := target.(pact.HasMigrations)
if !ok || len(hm.Migrations()) == 0 {
return fmt.Errorf("%w: %q", ErrNoMigrations, pluginID)
}
m, err := migrator(gdb, target.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
}
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
}