package parallelrun import ( "context" "errors" "fmt" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "gitea.perlbach24.de/scripte/nexarch/internal/cutoverorchestrator" "gitea.perlbach24.de/scripte/nexarch/internal/rollbackrun" ) var ( ErrNoApproval = errors.New("parallelrun: keine zweite freigabe zur abschaltung vorhanden") ErrDeadlineNotReached = errors.New("parallelrun: parallelbetriebsfrist noch nicht abgelaufen") ErrOpenDiscrepancies = errors.New("parallelrun: noch unbearbeitete diskrepanzen offen") ErrUnknownStatus = errors.New("parallelrun: unbekannter diskrepanz-status") ErrPeriodNotFound = errors.New("parallelrun: parallelbetriebsfrist nicht gefunden") ) type Period struct { ID int64 TenantID int64 DBName string CutoverAt time.Time Deadline time.Time CreatedAt time.Time } type Discrepancy struct { ID int64 PeriodID int64 ReportedAt time.Time Description string Status string ResolvedAt *time.Time ResolvedNote string } type ShutdownApproval struct { ID int64 PeriodID int64 ApprovedBy string ApprovedAt time.Time Note string } const ( StatusOpen = "open" StatusResolved = "resolved" StatusDeferred = "deferred" ) type Store struct { pool *pgxpool.Pool } func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} } // StartPeriod versetzt das Altsystem real in den Wartungsmodus // (schreibgeschuetzt, siehe rollbackrun.SetReadOnly) und persistiert die // Parallelbetriebsfrist mit Fristende. func (s *Store) StartPeriod(ctx context.Context, cluster *cutoverorchestrator.Cluster, tenantID int64, dbName string, cutoverAt time.Time, duration time.Duration) (int64, error) { if duration <= 0 { return 0, errors.New("parallelrun: frist muss positiv sein") } if err := rollbackrun.SetReadOnly(ctx, cluster, dbName, true); err != nil { return 0, fmt.Errorf("altsystem in wartungsmodus versetzen: %w", err) } deadline := cutoverAt.Add(duration) var id int64 err := s.pool.QueryRow(ctx, ` INSERT INTO parallel_run_period (tenant_id, db_name, cutover_at, deadline) VALUES ($1, $2, $3, $4) RETURNING id `, tenantID, dbName, cutoverAt, deadline).Scan(&id) if err != nil { return 0, fmt.Errorf("parallelbetriebsfrist speichern: %w", err) } return id, nil } func (s *Store) Period(ctx context.Context, periodID int64) (Period, error) { var p Period err := s.pool.QueryRow(ctx, ` SELECT id, tenant_id, db_name, cutover_at, deadline, created_at FROM parallel_run_period WHERE id = $1 `, periodID).Scan(&p.ID, &p.TenantID, &p.DBName, &p.CutoverAt, &p.Deadline, &p.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { return Period{}, ErrPeriodNotFound } if err != nil { return Period{}, fmt.Errorf("parallelbetriebsfrist lesen: %w", err) } return p, nil } func (s *Store) ReportDiscrepancy(ctx context.Context, periodID int64, description string) (int64, error) { if description == "" { return 0, errors.New("parallelrun: beschreibung darf nicht leer sein") } var id int64 err := s.pool.QueryRow(ctx, ` INSERT INTO parallel_run_discrepancy (period_id, description) VALUES ($1, $2) RETURNING id `, periodID, description).Scan(&id) if err != nil { return 0, fmt.Errorf("diskrepanz melden: %w", err) } return id, nil } // ResolveDiscrepancy markiert eine Diskrepanz als bearbeitet oder bewusst // zurueckgestellt; beides zaehlt fuer Akzeptanzkriterium 3 als erledigt. func (s *Store) ResolveDiscrepancy(ctx context.Context, discrepancyID int64, status, note string) error { if status != StatusResolved && status != StatusDeferred { return ErrUnknownStatus } tag, err := s.pool.Exec(ctx, ` UPDATE parallel_run_discrepancy SET status = $2, resolved_at = now(), resolved_note = $3 WHERE id = $1 `, discrepancyID, status, note) if err != nil { return fmt.Errorf("diskrepanz aktualisieren: %w", err) } if tag.RowsAffected() == 0 { return errors.New("parallelrun: diskrepanz nicht gefunden") } return nil } func (s *Store) OpenDiscrepancyCount(ctx context.Context, periodID int64) (int, error) { var n int err := s.pool.QueryRow(ctx, ` SELECT count(*) FROM parallel_run_discrepancy WHERE period_id = $1 AND status = $2 `, periodID, StatusOpen).Scan(&n) if err != nil { return 0, fmt.Errorf("offene diskrepanzen zaehlen: %w", err) } return n, nil } // ApproveShutdown speichert die zweite, eigenstaendige schriftliche Freigabe // zur Abschaltung. Sie ist real nur moeglich, wenn die Frist abgelaufen ist // und keine unbearbeiteten Diskrepanzen mehr offen sind (Akzeptanzkriterium 2/3). func (s *Store) ApproveShutdown(ctx context.Context, periodID int64, approvedBy, note string, now time.Time) (int64, error) { if approvedBy == "" { return 0, errors.New("parallelrun: approvedBy darf nicht leer sein") } period, err := s.Period(ctx, periodID) if err != nil { return 0, err } if now.Before(period.Deadline) { return 0, ErrDeadlineNotReached } open, err := s.OpenDiscrepancyCount(ctx, periodID) if err != nil { return 0, err } if open > 0 { return 0, ErrOpenDiscrepancies } var id int64 err = s.pool.QueryRow(ctx, ` INSERT INTO parallel_run_shutdown_approval (period_id, approved_by, note) VALUES ($1, $2, $3) RETURNING id `, periodID, approvedBy, note).Scan(&id) if err != nil { return 0, fmt.Errorf("abschaltungsfreigabe speichern: %w", err) } return id, nil } func (s *Store) LatestShutdownApproval(ctx context.Context, periodID int64) (ShutdownApproval, error) { var a ShutdownApproval err := s.pool.QueryRow(ctx, ` SELECT id, period_id, approved_by, approved_at, coalesce(note, '') FROM parallel_run_shutdown_approval WHERE period_id = $1 ORDER BY approved_at DESC, id DESC LIMIT 1 `, periodID).Scan(&a.ID, &a.PeriodID, &a.ApprovedBy, &a.ApprovedAt, &a.Note) if errors.Is(err, pgx.ErrNoRows) { return ShutdownApproval{}, ErrNoApproval } if err != nil { return ShutdownApproval{}, fmt.Errorf("abschaltungsfreigabe lesen: %w", err) } return a, nil } // Shutdown sperrt das Altsystem endgueltig (weder Lese- noch Schreibzugriff // mehr moeglich) und verweigert dies real, solange Frist oder zweite // Freigabe fehlen -- unabhaengig davon, ob ApproveShutdown diese Reihenfolge // bereits erzwungen hat. func Shutdown(ctx context.Context, store *Store, cluster *cutoverorchestrator.Cluster, periodID int64, now time.Time) error { period, err := store.Period(ctx, periodID) if err != nil { return err } if now.Before(period.Deadline) { return ErrDeadlineNotReached } if _, err := store.LatestShutdownApproval(ctx, periodID); err != nil { return err } open, err := store.OpenDiscrepancyCount(ctx, periodID) if err != nil { return err } if open > 0 { return ErrOpenDiscrepancies } conn, err := pgx.Connect(ctx, cluster.AdminDSN()) if err != nil { return fmt.Errorf("verbindung fuer abschaltung: %w", err) } defer conn.Close(ctx) ident := pgx.Identifier{period.DBName}.Sanitize() if _, err := conn.Exec(ctx, "ALTER DATABASE "+ident+" WITH ALLOW_CONNECTIONS false"); err != nil { return fmt.Errorf("verbindungen sperren: %w", err) } if _, err := conn.Exec(ctx, `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1`, period.DBName); err != nil { return fmt.Errorf("bestehende verbindungen beenden: %w", err) } return nil }