nexarch/internal/parallelrun/parallelrun.go

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
}