diff --git a/dms/go.mod b/dms/go.mod index 30c565a1..d3cfd07d 100644 --- a/dms/go.mod +++ b/dms/go.mod @@ -5,7 +5,7 @@ go 1.25.0 toolchain go1.25.13 require ( - gitea.perlbach24.de/scripte/nexarch/common v0.1.0 + gitea.perlbach24.de/scripte/nexarch/common v0.2.0 github.com/aws/aws-sdk-go-v2 v1.45.1 github.com/aws/aws-sdk-go-v2/config v1.33.1 github.com/aws/aws-sdk-go-v2/credentials v1.20.1 diff --git a/dms/go.sum b/dms/go.sum index bee8a5c4..a7b9c870 100644 --- a/dms/go.sum +++ b/dms/go.sum @@ -1,7 +1,7 @@ filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= -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/aws/aws-sdk-go-v2 v1.45.1 h1:iIoG3NaLhV6UZpPXyPXlDj2I9oS8tV/nMcMnITCC6Ks= github.com/aws/aws-sdk-go-v2 v1.45.1/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.20 h1:GPRlPwz40I2B2VrBEASOA3Bi77NyeqejNLkifosX0rs= diff --git a/dms/internal/migrate/migrate.go b/dms/internal/migrate/migrate.go index e8cf2feb..13e16023 100644 --- a/dms/internal/migrate/migrate.go +++ b/dms/internal/migrate/migrate.go @@ -1,224 +1,27 @@ -// Package migrate implementiert FDN-02s Migrationsmechanik: versionierte, -// rueckrollbare SQL-Migrationsdateien (kein ORM), mit einer -// dms_schema_migrations-Tabelle als Fortschrittsspeicher — dasselbe Prinzip wie -// NEXARCH Core (internal/migrate), hier eigenstaendig implementiert, da DMS -// ein eigenes Go-Modul ist und Cores internal/-Pakete 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_documents" - UpSQL string - DownSQL string -} +type Migration = cmigrate.Migration -// Load liest alle *.up.sql/*.down.sql-Paare aus dir, sortiert nach -// Dateiname (Akzeptanzkriterium: Migrationen sind versioniert). -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 (FDN-13). const TrackingTable = "dms_schema_migrations" -const legacyTrackingTable = "schema_migrations" +var migrator = cmigrate.MustNew(TrackingTable) -// ensureTrackingTable legt TrackingTable an. Beim erstmaligen Anlegen wird -// ein vorhandener Bestand aus der frueheren, unqualifizierten -// schema_migrations uebernommen (nur Versionen, die dieses Modul kennt, und -// nur wenn dessen Basisversion dort steht — eine schema_migrations eines -// anderen Moduls oder eines anderen Schemas wird weder gelesen noch -// veraendert). Die alte Tabelle -// bleibt unangetastet. -func ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) error { - tx, err := pool.Begin(ctx) - if err != nil { - return fmt.Errorf("%s anlegen: %w", TrackingTable, err) - } - defer func() { _ = tx.Rollback(ctx) }() +func Load(dir string) ([]Migration, error) { return cmigrate.Load(dir) } - // Serialisiert gleichzeitige Erstanlage/Uebernahme (mehrere App-Instanzen). - if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('dms_schema_migrations'))`); err != nil { - return fmt.Errorf("%s anlegen: %w", TrackingTable, err) - } - - var existed bool - if err := tx.QueryRow(ctx, `SELECT to_regclass(current_schema() || '.' || $1::text) IS NOT NULL`, TrackingTable).Scan(&existed); err != nil { - return fmt.Errorf("%s anlegen: %w", TrackingTable, err) - } - if existed { - return tx.Commit(ctx) - } - - if _, err := tx.Exec(ctx, ` - CREATE TABLE `+TrackingTable+` ( - version TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() - ) - `); err != nil { - return fmt.Errorf("%s anlegen: %w", TrackingTable, err) - } - - if len(migrations) > 0 { - var legacy bool - if err := tx.QueryRow(ctx, `SELECT to_regclass(current_schema() || '.' || $1::text) IS NOT NULL`, legacyTrackingTable).Scan(&legacy); err != nil { - return fmt.Errorf("%s pruefen: %w", legacyTrackingTable, err) - } - if legacy { - versions := make([]string, len(migrations)) - for i, m := range migrations { - versions[i] = m.Version - } - var hasBase bool - if err := tx.QueryRow(ctx, - `SELECT EXISTS (SELECT 1 FROM `+legacyTrackingTable+` WHERE version = $1)`, versions[0], - ).Scan(&hasBase); err != nil { - return fmt.Errorf("%s lesen: %w", legacyTrackingTable, err) - } - if hasBase { - if _, err := tx.Exec(ctx, ` - INSERT INTO `+TrackingTable+` (version, applied_at) - SELECT version, applied_at FROM `+legacyTrackingTable+` WHERE version = ANY($1) - `, versions); err != nil { - return fmt.Errorf("bestand aus %s uebernehmen: %w", legacyTrackingTable, 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 -// (Akzeptanzkriterium 2: vorwaerts ausfuehrbar) — bereits angewendete -// werden uebersprungen, damit Up auf einer leeren UND auf einer bestehenden -// DB funktioniert (Pruefung 1). -func Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (applied []string, err error) { - if err := ensureTrackingTable(ctx, pool, migrations); 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 -// (Akzeptanzkriterium 2: rueckwaerts ausfuehrbar) 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, migrations); 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) }