Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysopsandClaude Sonnet 5
|
b08f6a49fc |
@@ -1,88 +0,0 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DefaultCacheTTL ist die dokumentierte Cache-Invalidierungszeit
|
||||
// (Akzeptanzkriterium 2 / Pruefung 2 in diesem Ticket bezieht sich auf die
|
||||
// Aenderungsnachvollziehbarkeit — die Cache-Frist selbst folgt demselben
|
||||
// Muster wie internal/flag.DefaultCacheTTL).
|
||||
const DefaultCacheTTL = 5 * time.Second
|
||||
|
||||
type cacheEntry struct {
|
||||
value Value
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
// Service ist die Leseseite mit Vorrangregel (Akzeptanzkriterium 1:
|
||||
// Tenant-Override vor Global-Default) und lokalem TTL-Cache.
|
||||
type Service struct {
|
||||
store *Store
|
||||
ttl time.Duration
|
||||
|
||||
mu sync.RWMutex
|
||||
cache map[string]cacheEntry // Schluessel: key + "\x00" + tenantSlug
|
||||
}
|
||||
|
||||
func NewService(store *Store, ttl time.Duration) *Service {
|
||||
if ttl <= 0 {
|
||||
ttl = DefaultCacheTTL
|
||||
}
|
||||
return &Service{store: store, ttl: ttl, cache: make(map[string]cacheEntry)}
|
||||
}
|
||||
|
||||
func cacheKey(key, tenantSlug string) string {
|
||||
return key + "\x00" + tenantSlug
|
||||
}
|
||||
|
||||
// Resolve liefert den Konfigurationswert fuer einen Tenant: ein
|
||||
// Tenant-spezifischer Override hat Vorrang vor dem globalen Default
|
||||
// (Akzeptanzkriterium 1 / Pruefung 1). tenantSlug == "" wertet nur den
|
||||
// globalen Wert aus.
|
||||
func (s *Service) Resolve(ctx context.Context, tenantSlug, key string) (Value, error) {
|
||||
ck := cacheKey(key, tenantSlug)
|
||||
|
||||
s.mu.RLock()
|
||||
entry, exists := s.cache[ck]
|
||||
fresh := exists && time.Now().Before(entry.expiresAt)
|
||||
s.mu.RUnlock()
|
||||
if fresh {
|
||||
return entry.value, nil
|
||||
}
|
||||
|
||||
v, err := s.resolveUncached(ctx, tenantSlug, key)
|
||||
if err != nil {
|
||||
return Value{}, err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.cache[ck] = cacheEntry{value: v, expiresAt: time.Now().Add(s.ttl)}
|
||||
s.mu.Unlock()
|
||||
return v, nil
|
||||
}
|
||||
|
||||
func (s *Service) resolveUncached(ctx context.Context, tenantSlug, key string) (Value, error) {
|
||||
if tenantSlug != "" {
|
||||
v, err := s.store.Get(ctx, key, tenantSlug)
|
||||
if err == nil {
|
||||
return v, nil
|
||||
}
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
return Value{}, err
|
||||
}
|
||||
}
|
||||
return s.store.Get(ctx, key, GlobalScope)
|
||||
}
|
||||
|
||||
// Invalidate erzwingt beim naechsten Resolve-Aufruf ein sofortiges Neuladen
|
||||
// fuer einen bestimmten (key, tenantSlug) statt auf den TTL-Ablauf zu warten
|
||||
// — analog internal/flag.Service.Invalidate.
|
||||
func (s *Service) Invalidate(key, tenantSlug string) {
|
||||
s.mu.Lock()
|
||||
delete(s.cache, cacheKey(key, tenantSlug))
|
||||
s.mu.Unlock()
|
||||
}
|
||||
@@ -1,114 +0,0 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: Tenant-Override hat Vorrang vor
|
||||
// Global-Default, automatisiert getestet.
|
||||
func TestService_TenantOverrideTakesPrecedenceOverGlobal(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Set(ctx, "test_precedence_key", GlobalScope, "global-wert"); err != nil {
|
||||
t.Fatalf("set global: %v", err)
|
||||
}
|
||||
if _, err := store.Set(ctx, "test_precedence_key", "test_acme", "tenant-wert"); err != nil {
|
||||
t.Fatalf("set tenant: %v", err)
|
||||
}
|
||||
|
||||
svc := NewService(store, time.Hour)
|
||||
|
||||
got, err := svc.Resolve(ctx, "test_acme", "test_precedence_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve mit override: %v", err)
|
||||
}
|
||||
if got.Value != "tenant-wert" {
|
||||
t.Fatalf("erwartet tenant-override, habe %q", got.Value)
|
||||
}
|
||||
|
||||
gotOther, err := svc.Resolve(ctx, "test_anderer_tenant", "test_precedence_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve ohne override: %v", err)
|
||||
}
|
||||
if gotOther.Value != "global-wert" {
|
||||
t.Fatalf("erwartet global-default fuer tenant ohne override, habe %q", gotOther.Value)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Cache-Invalidierung nach
|
||||
// Konfigurationsaenderung innerhalb dokumentierter Zeit gemessen.
|
||||
func TestService_CacheInvalidationTiming(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
const ttl = 150 * time.Millisecond
|
||||
if _, err := store.Set(ctx, "test_ttl_key", GlobalScope, "alt"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, ttl)
|
||||
|
||||
v, err := svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value != "alt" {
|
||||
t.Fatalf("erwartet 'alt', habe %q", v.Value)
|
||||
}
|
||||
|
||||
changedAt := time.Now()
|
||||
if _, err := store.Set(ctx, "test_ttl_key", GlobalScope, "neu"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
|
||||
v, err = svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve direkt nach aenderung: %v", err)
|
||||
}
|
||||
if v.Value != "alt" {
|
||||
t.Fatalf("cache haette den alten wert liefern sollen, habe %q", v.Value)
|
||||
}
|
||||
|
||||
deadline := changedAt.Add(ttl + 100*time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
v, err := svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value == "neu" {
|
||||
t.Logf("aenderung wurde nach %s wirksam (ziel: innerhalb %s + toleranz)", time.Since(changedAt), ttl)
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("aenderung wurde nicht innerhalb von %s wirksam", deadline.Sub(changedAt))
|
||||
}
|
||||
|
||||
func TestService_InvalidateForcesImmediateRefresh(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Set(ctx, "test_invalidate_key", GlobalScope, "alt"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, time.Hour)
|
||||
_, _ = svc.Resolve(ctx, "", "test_invalidate_key")
|
||||
|
||||
if _, err := store.Set(ctx, "test_invalidate_key", GlobalScope, "neu"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc.Invalidate("test_invalidate_key", "")
|
||||
|
||||
v, err := svc.Resolve(ctx, "", "test_invalidate_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value != "neu" {
|
||||
t.Fatalf("erwartet sofort sichtbaren neuen wert nach Invalidate, habe %q", v.Value)
|
||||
}
|
||||
}
|
||||
@@ -1,129 +0,0 @@
|
||||
// Package cfgservice implementiert Core CFG-01: den zentralen Dienst fuer
|
||||
// globale und tenant-spezifische Konfigurationswerte mit Versionierung und
|
||||
// Cache-Invalidierung. Andere Module lesen Konfiguration AUSSCHLIESSLICH
|
||||
// ueber dieses Paket (Akzeptanzkriterium 3), niemals ueber eigene Tabellen.
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// GlobalScope ist der reservierte Scope-Wert fuer globale Defaults — jeder
|
||||
// andere Scope-Wert ist ein Tenant-Slug (Akzeptanzkriterium 1).
|
||||
const GlobalScope = "global"
|
||||
|
||||
var ErrNotFound = errors.New("cfgservice: kein wert fuer diesen key gefunden")
|
||||
|
||||
type Value struct {
|
||||
Key string
|
||||
Scope string
|
||||
Value string
|
||||
Version int
|
||||
}
|
||||
|
||||
type HistoryEntry struct {
|
||||
Key string
|
||||
Scope string
|
||||
Value string
|
||||
Version int
|
||||
}
|
||||
|
||||
// Store ist die Schreib-/Verwaltungsseite. Set schreibt IMMER sowohl den
|
||||
// aktuellen Stand (config_values) als auch einen Historieneintrag
|
||||
// (config_value_history) in derselben Transaktion — eine Aenderung ohne
|
||||
// Versionshistorie ist strukturell ausgeschlossen (Akzeptanzkriterium 2).
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// Set schreibt einen neuen Wert fuer (key, scope) und erhoeht die Version um 1
|
||||
// (Version 1 bei erstmaligem Setzen).
|
||||
func (s *Store) Set(ctx context.Context, key, scope, value string) (Value, error) {
|
||||
if scope == "" {
|
||||
return Value{}, errors.New("cfgservice: scope darf nicht leer sein")
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return Value{}, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
var currentVersion int
|
||||
err = tx.QueryRow(ctx, `SELECT version FROM config_values WHERE key = $1 AND scope = $2`, key, scope).Scan(¤tVersion)
|
||||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return Value{}, fmt.Errorf("aktuelle version lesen: %w", err)
|
||||
}
|
||||
newVersion := currentVersion + 1
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO config_values (key, scope, value, version, updated_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
ON CONFLICT (key, scope) DO UPDATE SET value = $3, version = $4, updated_at = now()
|
||||
`, key, scope, value, newVersion); err != nil {
|
||||
return Value{}, fmt.Errorf("wert speichern: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO config_value_history (key, scope, value, version, changed_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
`, key, scope, value, newVersion); err != nil {
|
||||
return Value{}, fmt.Errorf("historie schreiben: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return Value{}, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
|
||||
return Value{Key: key, Scope: scope, Value: value, Version: newVersion}, nil
|
||||
}
|
||||
|
||||
// Get liefert den Wert fuer GENAU EINEN Scope (kein Vorrang-Fallback) — die
|
||||
// Vorrangregel (Tenant vor Global) lebt bewusst in Service.Resolve, damit
|
||||
// Store rein CRUD bleibt.
|
||||
func (s *Store) Get(ctx context.Context, key, scope string) (Value, error) {
|
||||
var v Value
|
||||
v.Key, v.Scope = key, scope
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT value, version FROM config_values WHERE key = $1 AND scope = $2
|
||||
`, key, scope).Scan(&v.Value, &v.Version)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Value{}, ErrNotFound
|
||||
}
|
||||
return Value{}, fmt.Errorf("wert lesen: %w", err)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// History liefert die vollstaendige Versionshistorie eines (key, scope) in
|
||||
// aufsteigender Reihenfolge (Akzeptanzkriterium 2 / Pruefung 3).
|
||||
func (s *Store) History(ctx context.Context, key, scope string) ([]HistoryEntry, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT key, scope, value, version FROM config_value_history
|
||||
WHERE key = $1 AND scope = $2 ORDER BY version
|
||||
`, key, scope)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("historie abfragen: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []HistoryEntry
|
||||
for rows.Next() {
|
||||
var h HistoryEntry
|
||||
if err := rows.Scan(&h.Key, &h.Scope, &h.Value, &h.Version); err != nil {
|
||||
return nil, fmt.Errorf("historieneintrag lesen: %w", err)
|
||||
}
|
||||
out = append(out, h)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -1,119 +0,0 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupStoreTest(t *testing.T) (*Store, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS config_values (
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL CHECK (scope <> ''),
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (key, scope)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS config_value_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM config_value_history WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM config_values WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
||||
pool.Close()
|
||||
}
|
||||
return NewStore(pool), cleanup
|
||||
}
|
||||
|
||||
func TestStore_SetIncrementsVersion(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
v1, err := store.Set(ctx, "test_key", GlobalScope, "erster-wert")
|
||||
if err != nil {
|
||||
t.Fatalf("set 1: %v", err)
|
||||
}
|
||||
if v1.Version != 1 {
|
||||
t.Fatalf("erwartet version 1, habe %d", v1.Version)
|
||||
}
|
||||
|
||||
v2, err := store.Set(ctx, "test_key", GlobalScope, "zweiter-wert")
|
||||
if err != nil {
|
||||
t.Fatalf("set 2: %v", err)
|
||||
}
|
||||
if v2.Version != 2 {
|
||||
t.Fatalf("erwartet version 2, habe %d", v2.Version)
|
||||
}
|
||||
|
||||
got, err := store.Get(ctx, "test_key", GlobalScope)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.Value != "zweiter-wert" || got.Version != 2 {
|
||||
t.Fatalf("aktueller wert unerwartet: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 3: Versionierungshistorie ueber mehrere
|
||||
// Aenderungen hinweg nachvollzogen.
|
||||
func TestStore_HistoryTracksAllChanges(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
values := []string{"v1", "v2", "v3"}
|
||||
for _, v := range values {
|
||||
if _, err := store.Set(ctx, "test_history_key", GlobalScope, v); err != nil {
|
||||
t.Fatalf("set %q: %v", v, err)
|
||||
}
|
||||
}
|
||||
|
||||
history, err := store.History(ctx, "test_history_key", GlobalScope)
|
||||
if err != nil {
|
||||
t.Fatalf("history: %v", err)
|
||||
}
|
||||
if len(history) != 3 {
|
||||
t.Fatalf("erwartet 3 historieneintraege, habe %d", len(history))
|
||||
}
|
||||
for i, h := range history {
|
||||
if h.Version != i+1 || h.Value != values[i] {
|
||||
t.Fatalf("historieneintrag[%d] unerwartet: %+v", i, h)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStore_GetUnknownKeyReturnsNotFound(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Get(ctx, "test_nie_gesetzt", GlobalScope); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("erwartet ErrNotFound, habe %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,211 @@
|
||||
// Package migrate implementiert Core TEN-07: das automatisierte, pro Tenant
|
||||
// fehlerisolierte Ausrollen von SQL-Migrationen ueber alle registrierten
|
||||
// Mandanten-Datenbanken (Modell C — jede Migration muss N-mal statt einmal
|
||||
// laufen, siehe TEN-01).
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/tenant"
|
||||
)
|
||||
|
||||
const upSuffix = ".up.sql"
|
||||
|
||||
type Migration struct {
|
||||
Version string
|
||||
SQL string
|
||||
}
|
||||
|
||||
// LoadMigrations liest alle *.up.sql-Dateien aus dir und sortiert sie nach
|
||||
// Dateiname (die bestehende Namenskonvention 0001_..., 0002_... aus TEN-01
|
||||
// sorgt fuer eine stabile Reihenfolge).
|
||||
func LoadMigrations(dir string) ([]Migration, error) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("migrationsverzeichnis lesen: %w", err)
|
||||
}
|
||||
|
||||
var names []string
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() && strings.HasSuffix(e.Name(), upSuffix) {
|
||||
names = append(names, e.Name())
|
||||
}
|
||||
}
|
||||
sort.Strings(names)
|
||||
|
||||
migrations := make([]Migration, 0, len(names))
|
||||
for _, name := range names {
|
||||
content, err := os.ReadFile(filepath.Join(dir, name))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("migration %q lesen: %w", name, err)
|
||||
}
|
||||
version := strings.TrimSuffix(name, upSuffix)
|
||||
migrations = append(migrations, Migration{Version: version, SQL: string(content)})
|
||||
}
|
||||
return migrations, nil
|
||||
}
|
||||
|
||||
// MigrationStatus ist der pro Tenant und Version nachvollziehbare Stand
|
||||
// (Akzeptanzkriterium 3): welche Version, wann zuletzt versucht, Erfolg oder
|
||||
// Fehler.
|
||||
type MigrationStatus struct {
|
||||
Version string
|
||||
AppliedAt time.Time
|
||||
Success bool
|
||||
Error string
|
||||
}
|
||||
|
||||
// TenantResult fasst das Ergebnis eines Rollout-Versuchs fuer EINEN Tenant
|
||||
// zusammen — wird von Orchestrator.RolloutAll pro Tenant gesammelt, damit ein
|
||||
// Fehlschlag bei einem Mandanten die anderen nicht blockiert (Akzeptanzkriterium 2).
|
||||
type TenantResult struct {
|
||||
TenantSlug string
|
||||
Applied []string // erfolgreich in diesem Lauf angewendete Versionen
|
||||
FailedAt string // Version, bei der abgebrochen wurde; leer wenn kein Fehlschlag
|
||||
Err error
|
||||
}
|
||||
|
||||
// Orchestrator rollt Migrationen ueber alle in der Registry gefuehrten
|
||||
// Mandanten aus. Baut bewusst NICHT auf TEN-06 (Router) auf — Migrations-
|
||||
// Rollouts sind seltene Batch-Vorgaenge, kein Hot-Path, ein kurzlebiger Pool
|
||||
// pro Tenant und Lauf ist hier einfacher und unabhaengig von TEN-06 testbar.
|
||||
type Orchestrator struct {
|
||||
registry *tenant.Registry
|
||||
}
|
||||
|
||||
func NewOrchestrator(registry *tenant.Registry) *Orchestrator {
|
||||
return &Orchestrator{registry: registry}
|
||||
}
|
||||
|
||||
const ensureTableSQL = `
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version TEXT PRIMARY KEY,
|
||||
applied_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
success BOOLEAN NOT NULL,
|
||||
error TEXT
|
||||
);`
|
||||
|
||||
// RolloutAll wendet migrations auf JEDE registrierte Tenant-Datenbank an.
|
||||
// Jeder Tenant laeuft unabhaengig — ein Fehlschlag bei einem Mandanten wird
|
||||
// im jeweiligen TenantResult festgehalten und blockiert die uebrigen nicht
|
||||
// (Akzeptanzkriterium 1 + 2).
|
||||
func (o *Orchestrator) RolloutAll(ctx context.Context, migrations []Migration) ([]TenantResult, error) {
|
||||
tenants, err := o.registry.List(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("tenants fuer rollout auflisten: %w", err)
|
||||
}
|
||||
|
||||
results := make([]TenantResult, 0, len(tenants))
|
||||
for _, t := range tenants {
|
||||
results = append(results, o.rolloutForTenant(ctx, t, migrations))
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func (o *Orchestrator) rolloutForTenant(ctx context.Context, t tenant.Tenant, migrations []Migration) TenantResult {
|
||||
result := TenantResult{TenantSlug: t.Slug}
|
||||
|
||||
pool, err := pgxpool.New(ctx, t.DBDSN)
|
||||
if err != nil {
|
||||
result.Err = fmt.Errorf("verbindung zu tenant %q: %w", t.Slug, err)
|
||||
return result
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
if _, err := pool.Exec(ctx, ensureTableSQL); err != nil {
|
||||
result.Err = fmt.Errorf("schema_migrations anlegen fuer tenant %q: %w", t.Slug, err)
|
||||
return result
|
||||
}
|
||||
|
||||
for _, m := range migrations {
|
||||
alreadyApplied, err := isAlreadySuccessful(ctx, pool, m.Version)
|
||||
if err != nil {
|
||||
result.Err = fmt.Errorf("migrationsstand lesen fuer tenant %q, version %q: %w", t.Slug, m.Version, err)
|
||||
return result
|
||||
}
|
||||
if alreadyApplied {
|
||||
continue // Akzeptanzkriterium 3 / Pruefung 3: keine erneute Anwendung.
|
||||
}
|
||||
|
||||
_, execErr := pool.Exec(ctx, m.SQL)
|
||||
if execErr != nil {
|
||||
recordAttempt(ctx, pool, m.Version, false, execErr.Error())
|
||||
result.FailedAt = m.Version
|
||||
result.Err = fmt.Errorf("migration %q fuer tenant %q fehlgeschlagen: %w", m.Version, t.Slug, execErr)
|
||||
return result // spaetere Migrationen bauen typischerweise auf dieser auf — Abbruch NUR fuer diesen Tenant.
|
||||
}
|
||||
|
||||
recordAttempt(ctx, pool, m.Version, true, "")
|
||||
result.Applied = append(result.Applied, m.Version)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
func isAlreadySuccessful(ctx context.Context, pool *pgxpool.Pool, version string) (bool, error) {
|
||||
var success bool
|
||||
err := pool.QueryRow(ctx, `SELECT success FROM schema_migrations WHERE version = $1`, version).Scan(&success)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return success, nil
|
||||
}
|
||||
|
||||
func recordAttempt(ctx context.Context, pool *pgxpool.Pool, version string, success bool, errMsg string) {
|
||||
var errVal *string
|
||||
if errMsg != "" {
|
||||
errVal = &errMsg
|
||||
}
|
||||
_, _ = pool.Exec(ctx, `
|
||||
INSERT INTO schema_migrations (version, applied_at, success, error)
|
||||
VALUES ($1, now(), $2, $3)
|
||||
ON CONFLICT (version) DO UPDATE SET applied_at = now(), success = $2, error = $3
|
||||
`, version, success, errVal)
|
||||
}
|
||||
|
||||
// Status liefert den Migrationsstand eines einzelnen Tenants (Akzeptanzkriterium 3).
|
||||
func (o *Orchestrator) Status(ctx context.Context, tenantSlug string) ([]MigrationStatus, error) {
|
||||
t, err := o.registry.GetBySlug(ctx, tenantSlug)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("tenant %q nicht gefunden: %w", tenantSlug, err)
|
||||
}
|
||||
|
||||
pool, err := pgxpool.New(ctx, t.DBDSN)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("verbindung zu tenant %q: %w", tenantSlug, err)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
rows, err := pool.Query(ctx, `
|
||||
SELECT version, applied_at, success, COALESCE(error, '')
|
||||
FROM schema_migrations ORDER BY version
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("migrationsstand abfragen: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []MigrationStatus
|
||||
for rows.Next() {
|
||||
var s MigrationStatus
|
||||
if err := rows.Scan(&s.Version, &s.AppliedAt, &s.Success, &s.Error); err != nil {
|
||||
return nil, fmt.Errorf("migrationsstand lesen: %w", err)
|
||||
}
|
||||
out = append(out, s)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,238 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/tenant"
|
||||
)
|
||||
|
||||
func TestLoadMigrations_SortsByFilename(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
writeFile(t, dir, "0002_second.up.sql", "SELECT 2;")
|
||||
writeFile(t, dir, "0001_first.up.sql", "SELECT 1;")
|
||||
writeFile(t, dir, "0001_first.down.sql", "SELECT 'ignored';") // muss ignoriert werden
|
||||
|
||||
migrations, err := LoadMigrations(dir)
|
||||
if err != nil {
|
||||
t.Fatalf("load: %v", err)
|
||||
}
|
||||
if len(migrations) != 2 {
|
||||
t.Fatalf("erwartet 2 migrationen, habe %d", len(migrations))
|
||||
}
|
||||
if migrations[0].Version != "0001_first" || migrations[1].Version != "0002_second" {
|
||||
t.Fatalf("unerwartete reihenfolge: %+v", migrations)
|
||||
}
|
||||
}
|
||||
|
||||
func writeFile(t *testing.T, dir, name, content string) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(dir+"/"+name, []byte(content), 0o644); err != nil {
|
||||
t.Fatalf("write %s: %v", name, err)
|
||||
}
|
||||
}
|
||||
|
||||
func setupOrchestratorTest(t *testing.T, tenantCount int) (*Orchestrator, []tenant.Tenant, *pgxpool.Pool, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
adminPool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("admin pool: %v", err)
|
||||
}
|
||||
registryPool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("registry pool: %v", err)
|
||||
}
|
||||
if _, err := registryPool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS tenants (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
slug TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
db_name TEXT NOT NULL UNIQUE,
|
||||
db_dsn TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)`); err != nil {
|
||||
t.Fatalf("registry-schema: %v", err)
|
||||
}
|
||||
|
||||
registry := tenant.NewRegistry(registryPool)
|
||||
dsnTemplate := strings.Replace(adminDSN, "/postgres?", "/%s?", 1)
|
||||
provisioner := tenant.NewProvisioner(adminPool, registry, dsnTemplate)
|
||||
|
||||
var tenants []tenant.Tenant
|
||||
var slugs []string
|
||||
for i := 0; i < tenantCount; i++ {
|
||||
slug := fmt.Sprintf("mig_t%d", i)
|
||||
slugs = append(slugs, slug)
|
||||
tn, err := provisioner.Provision(ctx, slug, slug)
|
||||
if err != nil {
|
||||
t.Fatalf("provision %s: %v", slug, err)
|
||||
}
|
||||
tenants = append(tenants, tn)
|
||||
}
|
||||
|
||||
orchestrator := NewOrchestrator(registry)
|
||||
|
||||
cleanup := func() {
|
||||
for _, slug := range slugs {
|
||||
_, _ = adminPool.Exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, "tenant_"+slug))
|
||||
}
|
||||
_, _ = registryPool.Exec(ctx, `DELETE FROM tenants WHERE slug = ANY($1)`, slugs)
|
||||
registryPool.Close()
|
||||
adminPool.Close()
|
||||
}
|
||||
return orchestrator, tenants, adminPool, cleanup
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + 2 + Pruefung 1: Rollout gegen mehrere Tenants, einer
|
||||
// davon absichtlich inkompatibel — die anderen laufen trotzdem durch.
|
||||
func TestRolloutAll_IsolatesFailurePerTenant(t *testing.T) {
|
||||
orchestrator, tenants, _, cleanup := setupOrchestratorTest(t, 3)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
migrations := []Migration{
|
||||
{Version: "0001_demo_a", SQL: "CREATE TABLE demo_a (id INT);"},
|
||||
{Version: "0002_demo_b", SQL: "CREATE TABLE demo_b (id INT);"},
|
||||
}
|
||||
|
||||
// tenants[1] absichtlich inkompatibel machen: demo_b existiert schon,
|
||||
// migration 0002 schlaegt dort mit "already exists" fehl.
|
||||
badTenant := tenants[1]
|
||||
badPool, err := pgxpool.New(ctx, badTenant.DBDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("connect bad tenant: %v", err)
|
||||
}
|
||||
if _, err := badPool.Exec(ctx, "CREATE TABLE demo_b (id INT);"); err != nil {
|
||||
t.Fatalf("inkompatiblen zustand vorbereiten: %v", err)
|
||||
}
|
||||
badPool.Close()
|
||||
|
||||
results, err := orchestrator.RolloutAll(ctx, migrations)
|
||||
if err != nil {
|
||||
t.Fatalf("rollout: %v", err)
|
||||
}
|
||||
if len(results) != 3 {
|
||||
t.Fatalf("erwartet 3 ergebnisse, habe %d", len(results))
|
||||
}
|
||||
|
||||
byslug := map[string]TenantResult{}
|
||||
for _, r := range results {
|
||||
byslug[r.TenantSlug] = r
|
||||
}
|
||||
|
||||
good0 := byslug[tenants[0].Slug]
|
||||
if good0.Err != nil || len(good0.Applied) != 2 {
|
||||
t.Fatalf("tenant[0] sollte beide migrationen erhalten, habe %+v", good0)
|
||||
}
|
||||
good2 := byslug[tenants[2].Slug]
|
||||
if good2.Err != nil || len(good2.Applied) != 2 {
|
||||
t.Fatalf("tenant[2] sollte beide migrationen erhalten, habe %+v", good2)
|
||||
}
|
||||
|
||||
bad := byslug[badTenant.Slug]
|
||||
if bad.Err == nil {
|
||||
t.Fatal("erwartet fehler fuer den inkompatiblen tenant")
|
||||
}
|
||||
if bad.FailedAt != "0002_demo_b" {
|
||||
t.Fatalf("failedAt = %q, want 0002_demo_b", bad.FailedAt)
|
||||
}
|
||||
if len(bad.Applied) != 1 || bad.Applied[0] != "0001_demo_a" {
|
||||
t.Fatalf("erwartet dass 0001_demo_a trotzdem erfolgreich war, habe %+v", bad.Applied)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 2: Migrationsstand-Abfrage liefert fuer
|
||||
// jeden Tenant den korrekten, unabhaengigen Stand.
|
||||
func TestStatus_ReflectsPerTenantState(t *testing.T) {
|
||||
orchestrator, tenants, _, cleanup := setupOrchestratorTest(t, 1)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
migrations := []Migration{
|
||||
{Version: "0001_ok", SQL: "CREATE TABLE ok_table (id INT);"},
|
||||
{Version: "0002_fail", SQL: "SELECT this_column_does_not_exist FROM ok_table;"},
|
||||
}
|
||||
|
||||
if _, err := orchestrator.RolloutAll(ctx, migrations); err != nil {
|
||||
t.Fatalf("rollout: %v", err)
|
||||
}
|
||||
|
||||
status, err := orchestrator.Status(ctx, tenants[0].Slug)
|
||||
if err != nil {
|
||||
t.Fatalf("status: %v", err)
|
||||
}
|
||||
if len(status) != 2 {
|
||||
t.Fatalf("erwartet 2 status-eintraege, habe %d", len(status))
|
||||
}
|
||||
if !status[0].Success || status[0].Version != "0001_ok" {
|
||||
t.Fatalf("status[0] unerwartet: %+v", status[0])
|
||||
}
|
||||
if status[1].Success || status[1].Version != "0002_fail" || status[1].Error == "" {
|
||||
t.Fatalf("status[1] sollte fehlgeschlagen sein mit fehlertext: %+v", status[1])
|
||||
}
|
||||
}
|
||||
|
||||
// Pruefung 3: wiederholter Rollout-Versuch wendet bereits erfolgreiche
|
||||
// Migrationen NICHT erneut an und kann die zuvor fehlgeschlagene nachholen,
|
||||
// sobald die Ursache behoben ist.
|
||||
func TestRolloutAll_RetryDoesNotReapplySuccessful(t *testing.T) {
|
||||
orchestrator, tenants, _, cleanup := setupOrchestratorTest(t, 1)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
// 0002 schlaegt beim ersten Versuch fehl, weil demo_conflict schon
|
||||
// existiert (wir legen sie vorher an, um den Fehlschlag zu erzwingen).
|
||||
pool, err := pgxpool.New(ctx, tenants[0].DBDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("connect: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, "CREATE TABLE demo_conflict (id INT);"); err != nil {
|
||||
t.Fatalf("vorbedingung: %v", err)
|
||||
}
|
||||
|
||||
migrations := []Migration{
|
||||
{Version: "0001_ok", SQL: "CREATE TABLE demo_first (id INT);"}, // OHNE IF NOT EXISTS,
|
||||
// damit ein erneutes Anwenden nachweislich fehlschlagen wuerde.
|
||||
{Version: "0002_conflict", SQL: "CREATE TABLE demo_conflict (id INT);"},
|
||||
}
|
||||
|
||||
firstRun, err := orchestrator.RolloutAll(ctx, migrations)
|
||||
if err != nil {
|
||||
t.Fatalf("rollout 1: %v", err)
|
||||
}
|
||||
if firstRun[0].FailedAt != "0002_conflict" {
|
||||
t.Fatalf("erwartet fehlschlag bei 0002_conflict im ersten lauf, habe %+v", firstRun[0])
|
||||
}
|
||||
|
||||
// Ursache beheben.
|
||||
if _, err := pool.Exec(ctx, "DROP TABLE demo_conflict;"); err != nil {
|
||||
t.Fatalf("ursache beheben: %v", err)
|
||||
}
|
||||
pool.Close()
|
||||
|
||||
secondRun, err := orchestrator.RolloutAll(ctx, migrations)
|
||||
if err != nil {
|
||||
t.Fatalf("rollout 2: %v", err)
|
||||
}
|
||||
// Waere 0001_ok erneut angewendet worden ("CREATE TABLE demo_first" ohne
|
||||
// IF NOT EXISTS), haette das einen Fehler erzeugt statt eines sauberen
|
||||
// Applied-Eintrags fuer 0002_conflict.
|
||||
if secondRun[0].Err != nil {
|
||||
t.Fatalf("zweiter lauf sollte fehlerfrei sein, habe %+v", secondRun[0])
|
||||
}
|
||||
if len(secondRun[0].Applied) != 1 || secondRun[0].Applied[0] != "0002_conflict" {
|
||||
t.Fatalf("erwartet nur 0002_conflict im zweiten lauf angewendet, habe %+v", secondRun[0].Applied)
|
||||
}
|
||||
}
|
||||
@@ -1,2 +0,0 @@
|
||||
DROP TABLE IF EXISTS config_value_history;
|
||||
DROP TABLE IF EXISTS config_values;
|
||||
@@ -1,23 +0,0 @@
|
||||
-- Zentraler Konfigurationsdienst (CFG-01, siehe core-kanban/tickets/CFG-01.md).
|
||||
-- scope = 'global' fuer globale Defaults, sonst der Tenant-Slug. config_values
|
||||
-- haelt den AKTUELLEN Stand je (key, scope); config_value_history haelt JEDE
|
||||
-- Aenderung fest (Akzeptanzkriterium 2: versioniert nachvollziehbar).
|
||||
CREATE TABLE config_values (
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL CHECK (scope <> ''),
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (key, scope)
|
||||
);
|
||||
|
||||
CREATE TABLE config_value_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX config_value_history_key_scope_idx ON config_value_history (key, scope, version);
|
||||
@@ -1,26 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
# Setzt die nexarch-Testumgebung zurueck: loescht die geteilte
|
||||
# Registry-Tabelle "tenants" in der postgres-Wartungsdatenbank sowie alle
|
||||
# tenant_*-Datenbanken. Noetig, weil verschiedene Feature-Branches
|
||||
# unterschiedliche Registry-Schemata erwarten, aber dieselbe physische
|
||||
# Postgres-Instanz auf dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
|
||||
#
|
||||
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/reset-test-env.sh
|
||||
set -euo pipefail
|
||||
|
||||
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
||||
ROLE="nexarch_test"
|
||||
|
||||
export PGPASSWORD="$PASS"
|
||||
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS audit_events CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS config_value_history CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS config_values CASCADE;"
|
||||
|
||||
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
|
||||
for db in $dbs; do
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS \"${db}\";"
|
||||
done
|
||||
|
||||
echo "Testumgebung zurueckgesetzt: registry-tabelle + $(echo "$dbs" | grep -c . || true) tenant-datenbank(en) entfernt."
|
||||
@@ -1,24 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
# Ein-Kommando-Pruefung fuer den aktuellen Code-Stand auf dem Testhost:
|
||||
# Registry+Tenant-DBs zuruecksetzen, dann build/vet/test in einem Rutsch.
|
||||
# -p 1 ist Pflicht, da mehrere Pakete dieselbe physische Registry-Tabelle auf
|
||||
# dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
|
||||
#
|
||||
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/run-checks.sh
|
||||
set -euo pipefail
|
||||
|
||||
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
||||
cd "$(dirname "$0")/.."
|
||||
|
||||
NEXARCH_TEST_DB_PASSWORD="$PASS" bash scripts/reset-test-env.sh
|
||||
|
||||
export TEST_ADMIN_DSN="postgresql://nexarch_test:${PASS}@localhost:5432/postgres?sslmode=disable"
|
||||
|
||||
echo "== go build =="
|
||||
go build ./...
|
||||
|
||||
echo "== go vet =="
|
||||
go vet ./...
|
||||
|
||||
echo "== go test (-p 1) =="
|
||||
go test ./... -p 1 -count=1
|
||||
Reference in New Issue
Block a user