diff --git a/workflow/go.mod b/workflow/go.mod index 8ec87be2..3acdf23e 100644 --- a/workflow/go.mod +++ b/workflow/go.mod @@ -3,7 +3,7 @@ module gitea.perlbach24.de/scripte/nexarch/workflow go 1.25.0 require ( - gitea.perlbach24.de/scripte/nexarch/common v0.1.0 + gitea.perlbach24.de/scripte/nexarch/common v0.2.0 github.com/jackc/pgx/v5 v5.9.2 ) diff --git a/workflow/go.sum b/workflow/go.sum index a5cc3c17..9ca36777 100644 --- a/workflow/go.sum +++ b/workflow/go.sum @@ -1,5 +1,5 @@ -gitea.perlbach24.de/scripte/nexarch/common v0.1.0 h1:biNzFr2LOMZC7b1leY3smQtrMmSMp6IW80OvR4LSjLY= -gitea.perlbach24.de/scripte/nexarch/common v0.1.0/go.mod h1:CiEImG++QMt1xZqSMeDJhPiB7sfWTLZOwjS6OY/Iguc= +gitea.perlbach24.de/scripte/nexarch/common v0.2.0 h1:gcOjoOvUkT0Zmqq3m7WQp1MfXSgmzrGTF2E/+NkO2bw= +gitea.perlbach24.de/scripte/nexarch/common v0.2.0/go.mod h1:CiEImG++QMt1xZqSMeDJhPiB7sfWTLZOwjS6OY/Iguc= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/workflow/internal/migrate/migrate.go b/workflow/internal/migrate/migrate.go index 92526fd5..d70e9a57 100644 --- a/workflow/internal/migrate/migrate.go +++ b/workflow/internal/migrate/migrate.go @@ -1,177 +1,27 @@ -// Package migrate implementiert Workflows Migrationsmechanik: versionierte, -// rueckrollbare SQL-Migrationsdateien (kein ORM), mit einer -// workflow_schema_migrations-Tabelle als Fortschrittsspeicher — dasselbe -// Prinzip wie NEXARCH Core und DMS (internal/migrate), hier eigenstaendig -// implementiert, da Workflow ein eigenes Go-Modul ist und die internal/- -// Pakete der anderen Module nicht importieren kann. +// Package migrate bindet die gemeinsame Migrationslogik aus common an den +// modulspezifischen Fortschrittstabellennamen. package migrate import ( "context" - "fmt" - "os" - "path/filepath" - "sort" - "strings" "github.com/jackc/pgx/v5/pgxpool" + + cmigrate "gitea.perlbach24.de/scripte/nexarch/common/migrate" ) -// Migration ist eine einzelne versionierte Migrationsdatei. -type Migration struct { - Version string // Dateiname ohne .up.sql/.down.sql, z.B. "0001_workflow_instances" - UpSQL string - DownSQL string -} +type Migration = cmigrate.Migration -// Load liest alle *.up.sql/*.down.sql-Paare aus dir, sortiert nach -// Dateiname. -func Load(dir string) ([]Migration, error) { - entries, err := os.ReadDir(dir) - if err != nil { - return nil, fmt.Errorf("migrationsverzeichnis %q lesen: %w", dir, err) - } - var versions []string - for _, e := range entries { - if e.IsDir() || !strings.HasSuffix(e.Name(), ".up.sql") { - continue - } - versions = append(versions, strings.TrimSuffix(e.Name(), ".up.sql")) - } - sort.Strings(versions) - - migrations := make([]Migration, 0, len(versions)) - for _, v := range versions { - up, err := os.ReadFile(filepath.Join(dir, v+".up.sql")) - if err != nil { - return nil, fmt.Errorf("migration %q: up.sql lesen: %w", v, err) - } - down, err := os.ReadFile(filepath.Join(dir, v+".down.sql")) - if err != nil { - return nil, fmt.Errorf("migration %q: down.sql lesen (jede migration braucht ein rollback): %w", v, err) - } - migrations = append(migrations, Migration{Version: v, UpSQL: string(up), DownSQL: string(down)}) - } - return migrations, nil -} - -// TrackingTable ist modulspezifisch benannt, damit sich Module in -// derselben Datenbank/demselben Schema nicht gegenseitig den -// Migrationsfortschritt ueberschreiben. const TrackingTable = "workflow_schema_migrations" -func ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool) error { - tx, err := pool.Begin(ctx) - if err != nil { - return fmt.Errorf("%s anlegen: %w", TrackingTable, err) - } - defer func() { _ = tx.Rollback(ctx) }() +var migrator = cmigrate.MustNew(TrackingTable) - if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('workflow_schema_migrations'))`); err != nil { - return fmt.Errorf("%s anlegen: %w", TrackingTable, err) - } +func Load(dir string) ([]Migration, error) { return cmigrate.Load(dir) } - if _, err := tx.Exec(ctx, ` - CREATE TABLE IF NOT EXISTS `+TrackingTable+` ( - version TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() - ) - `); err != nil { - return fmt.Errorf("%s anlegen: %w", TrackingTable, err) - } - return tx.Commit(ctx) +func Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) ([]string, error) { + return migrator.Up(ctx, pool, migrations) } -func appliedVersions(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) { - rows, err := pool.Query(ctx, `SELECT version FROM `+TrackingTable) - if err != nil { - return nil, fmt.Errorf("angewendete migrationen lesen: %w", err) - } - defer rows.Close() - - applied := map[string]bool{} - for rows.Next() { - var v string - if err := rows.Scan(&v); err != nil { - return nil, fmt.Errorf("migrationsversion lesen: %w", err) - } - applied[v] = true - } - return applied, rows.Err() -} - -// Up wendet alle noch nicht angewendeten Migrationen in Reihenfolge an; -// bereits angewendete werden uebersprungen (funktioniert auf leerer UND -// bestehender DB). -func Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (applied []string, err error) { - if err := ensureTrackingTable(ctx, pool); err != nil { - return nil, err - } - already, err := appliedVersions(ctx, pool) - if err != nil { - return nil, err - } - - for _, m := range migrations { - if already[m.Version] { - continue - } - tx, err := pool.Begin(ctx) - if err != nil { - return applied, fmt.Errorf("transaktion fuer %q starten: %w", m.Version, err) - } - if _, err := tx.Exec(ctx, m.UpSQL); err != nil { - _ = tx.Rollback(ctx) - return applied, fmt.Errorf("migration %q anwenden: %w", m.Version, err) - } - if _, err := tx.Exec(ctx, `INSERT INTO `+TrackingTable+` (version) VALUES ($1)`, m.Version); err != nil { - _ = tx.Rollback(ctx) - return applied, fmt.Errorf("migration %q als angewendet markieren: %w", m.Version, err) - } - if err := tx.Commit(ctx); err != nil { - return applied, fmt.Errorf("migration %q committen: %w", m.Version, err) - } - applied = append(applied, m.Version) - } - return applied, nil -} - -// DownOne macht die zuletzt angewendete Migration rueckgaengig und liefert -// deren Version, oder "" falls keine Migration angewendet war. -func DownOne(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (version string, err error) { - if err := ensureTrackingTable(ctx, pool); err != nil { - return "", err - } - already, err := appliedVersions(ctx, pool) - if err != nil { - return "", err - } - - var last *Migration - for i := len(migrations) - 1; i >= 0; i-- { - if already[migrations[i].Version] { - last = &migrations[i] - break - } - } - if last == nil { - return "", nil - } - - tx, err := pool.Begin(ctx) - if err != nil { - return "", fmt.Errorf("transaktion fuer rollback von %q starten: %w", last.Version, err) - } - if _, err := tx.Exec(ctx, last.DownSQL); err != nil { - _ = tx.Rollback(ctx) - return "", fmt.Errorf("migration %q zurueckrollen: %w", last.Version, err) - } - if _, err := tx.Exec(ctx, `DELETE FROM `+TrackingTable+` WHERE version = $1`, last.Version); err != nil { - _ = tx.Rollback(ctx) - return "", fmt.Errorf("migration %q aus %s entfernen: %w", last.Version, TrackingTable, err) - } - if err := tx.Commit(ctx); err != nil { - return "", fmt.Errorf("rollback von %q committen: %w", last.Version, err) - } - return last.Version, nil +func DownOne(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (string, error) { + return migrator.DownOne(ctx, pool, migrations) }