nexarch/internal/rollbackrun/rollbackrun.go

package rollbackrun

import (
	"context"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/internal/cutoverorchestrator"
)

const legacyStateSchemaSQL = `
CREATE TABLE legacy_system_state (
    tenant_id        BIGINT PRIMARY KEY,
    maintenance_mode BOOLEAN NOT NULL DEFAULT false,
    nexarch_enabled  BOOLEAN NOT NULL DEFAULT false,
    updated_at       TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE legacy_writeprobe (
    id         BIGSERIAL PRIMARY KEY,
    note       TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
`

// BusinessTables werden fuer den Datenverlust-Vergleich herangezogen.
// legacy_system_state/legacy_writeprobe gehoeren nicht dazu: sie druecken
// den Wartungsmodus- bzw. Schreibtest-Zustand selbst aus, der sich durch
// den Rollback gewollt aendert.
var BusinessTables = []string{
	"tenants", "tenant_domains", "users", "storage_objects", "emails",
	"attachments", "email_attachments", "email_refs", "audit_log",
	"reconciliation_reports", "dsgvo_requests",
}

type Testbed struct {
	DBName   string
	Pool     *pgxpool.Pool
	TenantID int64
}

func Setup(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName string, tenantID int64, data cutoverorchestrator.ArchivmailData) (*Testbed, error) {
	if err := cluster.CreateDatabase(ctx, dbName); err != nil {
		return nil, err
	}
	pool, err := cluster.Connect(ctx, dbName)
	if err != nil {
		return nil, err
	}
	if _, err := cutoverorchestrator.SeedArchivmail(ctx, pool, data); err != nil {
		pool.Close()
		return nil, err
	}
	if _, err := pool.Exec(ctx, legacyStateSchemaSQL); err != nil {
		pool.Close()
		return nil, fmt.Errorf("legacy_system_state anlegen: %w", err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO legacy_system_state (tenant_id, maintenance_mode, nexarch_enabled) VALUES ($1, true, true)`, tenantID); err != nil {
		pool.Close()
		return nil, fmt.Errorf("wartungszustand setzen: %w", err)
	}
	if _, err := pool.Exec(ctx, "ALTER DATABASE "+pgx.Identifier{dbName}.Sanitize()+" SET default_transaction_read_only = on"); err != nil {
		pool.Close()
		return nil, fmt.Errorf("wartungsmodus aktivieren: %w", err)
	}
	pool.Close()

	roPool, err := cluster.Connect(ctx, dbName)
	if err != nil {
		return nil, fmt.Errorf("neuverbindung im wartungsmodus: %w", err)
	}
	return &Testbed{DBName: dbName, Pool: roPool, TenantID: tenantID}, nil
}

// SetReadOnly wird auch von internal/parallelrun fuer den
// Parallelbetrieb-Wartungsmodus wiederverwendet, statt die ALTER-DATABASE-
// Anweisung zu duplizieren.
func SetReadOnly(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName string, readOnly bool) error {
	conn, err := pgx.Connect(ctx, cluster.AdminDSN())
	if err != nil {
		return fmt.Errorf("verbindung fuer wartungsmodus-umschaltung: %w", err)
	}
	defer conn.Close(ctx)
	setting := "off"
	if readOnly {
		setting = "on"
	}
	if _, err := conn.Exec(ctx, "ALTER DATABASE "+pgx.Identifier{dbName}.Sanitize()+" SET default_transaction_read_only = "+setting); err != nil {
		return fmt.Errorf("wartungsmodus umschalten: %w", err)
	}
	return nil
}

func DropDatabase(ctx context.Context, cluster *cutoverorchestrator.Cluster, dbName string) error {
	conn, err := pgx.Connect(ctx, cluster.AdminDSN())
	if err != nil {
		return err
	}
	defer conn.Close(ctx)
	_, err = conn.Exec(ctx, "DROP DATABASE IF EXISTS "+pgx.Identifier{dbName}.Sanitize()+" WITH (FORCE)")
	return err
}

type tableSnapshot struct {
	count    int64
	checksum string
}

func snapshotTables(ctx context.Context, pool *pgxpool.Pool, tables []string) (map[string]tableSnapshot, error) {
	out := make(map[string]tableSnapshot, len(tables))
	for _, tbl := range tables {
		q := fmt.Sprintf(`SELECT count(*), coalesce(md5(string_agg(r, '' ORDER BY r)), '') FROM (SELECT (t.*)::text AS r FROM %s t) s`,
			pgx.Identifier{tbl}.Sanitize())
		var snap tableSnapshot
		if err := pool.QueryRow(ctx, q).Scan(&snap.count, &snap.checksum); err != nil {
			return nil, fmt.Errorf("snapshot tabelle %s: %w", tbl, err)
		}
		out[tbl] = snap
	}
	return out, nil
}

type Report struct {
	TenantID         int64
	Start            time.Time
	End              time.Time
	Duration         time.Duration
	WriteVerified    bool
	WriteProbeID     int64
	TablesCompared   []string
	DataLossDetected bool
	MismatchedTables []string
}

// Rollback hebt den Wartungsmodus des Altsystems auf und nimmt die
// NEXARCH-Freischaltung des Tenants zurueck. Die gemessene Dauer umfasst
// genau diese Operation (ohne die flankierenden Vorher-/Nachher-Pruefungen).
func Rollback(ctx context.Context, cluster *cutoverorchestrator.Cluster, tb *Testbed) (Report, error) {
	before, err := snapshotTables(ctx, tb.Pool, BusinessTables)
	if err != nil {
		return Report{}, err
	}

	var maint, nex bool
	if err := tb.Pool.QueryRow(ctx, `SELECT maintenance_mode, nexarch_enabled FROM legacy_system_state WHERE tenant_id = $1`, tb.TenantID).Scan(&maint, &nex); err != nil {
		return Report{}, fmt.Errorf("wartungszustand lesen: %w", err)
	}
	if !maint || !nex {
		return Report{}, fmt.Errorf("rollback-vorbedingung verletzt: wartungsmodus=%v freischaltung=%v", maint, nex)
	}

	start := time.Now()

	conn, err := tb.Pool.Acquire(ctx)
	if err != nil {
		return Report{}, fmt.Errorf("verbindung fuer wartungsmodus-aufhebung: %w", err)
	}
	// SET transaction_read_only gilt nur fuer die laufende Transaktion;
	// ohne den Override in derselben impliziten Transaktion wie die
	// ALTER-DATABASE-Anweisung wuerde diese als Schreibzugriff unter dem
	// noch aktiven default_transaction_read_only=on abgelehnt (SQLSTATE 25006).
	_, err = conn.Exec(ctx, "SET transaction_read_only = off; ALTER DATABASE "+pgx.Identifier{tb.DBName}.Sanitize()+" SET default_transaction_read_only = off;")
	conn.Release()
	if err != nil {
		return Report{}, fmt.Errorf("wartungsmodus aufheben: %w", err)
	}
	tb.Pool.Close()

	rwPool, err := cluster.Connect(ctx, tb.DBName)
	if err != nil {
		return Report{}, fmt.Errorf("neuverbindung nach rollback: %w", err)
	}
	tb.Pool = rwPool

	if _, err := tb.Pool.Exec(ctx, `UPDATE legacy_system_state SET maintenance_mode = false, nexarch_enabled = false, updated_at = now() WHERE tenant_id = $1`, tb.TenantID); err != nil {
		return Report{}, fmt.Errorf("freischaltung zuruecknehmen: %w", err)
	}

	var probeID int64
	if err := tb.Pool.QueryRow(ctx, `INSERT INTO legacy_writeprobe (note) VALUES ('rollback-schreibtest') RETURNING id`).Scan(&probeID); err != nil {
		return Report{}, fmt.Errorf("altsystem nach rollback nicht beschreibbar: %w", err)
	}

	end := time.Now()

	after, err := snapshotTables(ctx, tb.Pool, BusinessTables)
	if err != nil {
		return Report{}, err
	}

	var mismatches []string
	for _, tbl := range BusinessTables {
		if before[tbl] != after[tbl] {
			mismatches = append(mismatches, tbl)
		}
	}

	return Report{
		TenantID:         tb.TenantID,
		Start:            start,
		End:              end,
		Duration:         end.Sub(start),
		WriteVerified:    true,
		WriteProbeID:     probeID,
		TablesCompared:   append([]string{}, BusinessTables...),
		DataLossDetected: len(mismatches) > 0,
		MismatchedTables: mismatches,
	}, nil
}