munin/internal/store/speicher_postgres.go

package store

import (
	"context"
	"database/sql"
	"embed"
	"fmt"
	"log"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/stdlib"
)

// ── Verbindungen ─────────────────────────────────────────────────────────────
//
// PostgreSQL erlaubt nur eine feste Zahl gleichzeitiger Verbindungen
// (max_connections, Standard 100). Würde Munin mehr öffnen, lehnte PostgreSQL
// neue Verbindungen ab und Anfragen schlügen fehl. Deshalb gelten hier feste,
// kleine Pools, und Munin rechnet vor dem Start und vor jedem neuen Workspace
// nach, ob sie reichen (PruefeKapazitaet). Hintergrund und Rechnung:
// AUSBAUSTUFEN.md 4.10, „Verbindungen fest begrenzt“.
//
// Kleine Pools reichen, weil eine Verbindung nur für die Dauer einer Abfrage
// belegt ist — Millisekunden. Während ein Agent nachdenkt, hält Munin keine.
// Ist ein Pool voll, wartet die Anfrage in Go auf eine freie Verbindung (bis
// zum Ende ihres Kontexts), statt in PostgreSQL zu scheitern.
const (
	// pgLeseverbindungen je Workspace (und für die Kontroll-Datenbank).
	pgLeseverbindungen = 2

	// pgSchreibverbindungen je Workspace. Vorerst 1, also dasselbe Verhalten
	// wie unter SQLite: Schreibvorgänge eines Workspace laufen nacheinander.
	// Mehr erst, wenn die schreibenden Methoden gegen gleichzeitiges Schreiben
	// abgesichert sind (AUSBAUSTUFEN.md 4.10, Schritt 4) — sonst kann z.B.
	// PutEntry unter PostgreSQL Änderungen verlieren.
	pgSchreibverbindungen = 1

	// pgVerbindungenJeWorkspace ist, was ein geöffneter Workspace höchstens
	// belegt.
	pgVerbindungenJeWorkspace = pgLeseverbindungen + pgSchreibverbindungen

	// pgFesteVerbindungen: die Kontroll-Datenbank (wie ein Workspace) plus
	// eine Verwaltungsverbindung zum Anlegen und Löschen von Schemas.
	pgFesteVerbindungen = pgVerbindungenJeWorkspace + 1

	// pgReserve bleibt frei — für psql, Backups (pg_dump) und die
	// Verbindungen, die PostgreSQL selbst für Superuser zurückhält.
	pgReserve = 10

	// pgLeerlauf: Nach so langer Ruhe schließt ein Pool seine Verbindungen.
	// Ein Workspace, den gerade niemand benutzt, belegt dann keine.
	pgLeerlauf = 2 * time.Minute

	// Beim Start ist PostgreSQL vielleicht noch nicht bereit (der Container
	// fährt gerade hoch). Munin versucht es so oft, mit dieser Pause dazwischen.
	pgStartVersuche = 15
	pgStartPause    = 2 * time.Second
)

// ── Schemanamen ──────────────────────────────────────────────────────────────

// Alles von Munin liegt in Schemas mit dem Präfix "munin_" — auf einen Blick
// erkennbar (\dn munin_*), auch wenn in derselben Datenbank andere
// Anwendungen liegen. Begründung: AUSBAUSTUFEN.md 4.10, „Schemanamen aus der
// Workspace-ID“.
const (
	ControlSchema          = "munin_control"
	workspaceSchemaPraefix = "munin_ws_"
)

// erlaubteWorkspaceID passt auf IDs, wie NewID sie erzeugt (Kleinbuchstaben,
// Ziffern, Bindestrich). Alles andere wird abgelehnt, bevor es in einen
// Schemanamen wandert.
var erlaubteWorkspaceID = regexp.MustCompile(`^[a-z0-9-]+$`)

// WorkspaceSchema liefert den Schemanamen eines Workspace, z.B.
// "munin_ws_mue9bh3x_1c0e1120".
//
// Die ID statt des Namens, weil sich Namen ändern können und freier Text sind.
// "-" wird zu "_", weil ein Bindestrich in PostgreSQL überall Anführungszeichen
// erzwingen würde. Den lesbaren Namen trägt ein Kommentar am Schema (siehe
// nachDemBenennen).
func WorkspaceSchema(workspaceID string) (string, error) {
	if !erlaubteWorkspaceID.MatchString(workspaceID) {
		return "", fmt.Errorf("Workspace-ID %q ist als Schemaname nicht geeignet", workspaceID)
	}
	name := workspaceSchemaPraefix + strings.ReplaceAll(workspaceID, "-", "_")

	// PostgreSQL kürzt längere Namen stillschweigend auf 63 Byte. Zwei lange
	// IDs könnten dann im selben Schema landen — lieber gleich ablehnen.
	if len(name) > 63 {
		return "", fmt.Errorf("Workspace-ID %q ergibt einen zu langen Schemanamen", workspaceID)
	}
	return name, nil
}

// ── Speicher ─────────────────────────────────────────────────────────────────

// PostgresSpeicher legt die Kontroll-Datenbank und jeden Workspace als eigenes
// Schema in einer PostgreSQL-Datenbank ab.
//
// Jedes Schema bekommt eigene Pools, deren Verbindungen mit
// search_path=<schema> starten. Dadurch laufen dieselben Abfragen wie unter
// SQLite unverändert — "SELECT … FROM entries" trifft automatisch die Tabelle
// im richtigen Schema, und ein Workspace kann die Tabellen eines anderen gar
// nicht erst sehen.
type PostgresSpeicher struct {
	basis *pgx.ConnConfig // aus DATABASE_URL; je Schema kopiert und ergänzt
	admin *sql.DB         // eine Verbindung ohne search_path, für CREATE/DROP SCHEMA

	// controlSchema ist im Betrieb immer ControlSchema. Nur Tests setzen ein
	// eigenes, damit sie die Kontroll-Datenbank eines Entwicklers nicht
	// anfassen (siehe postgres_test.go).
	controlSchema string

	// hinweise sammelt beim Start, was nicht zum Abbruch reicht, aber auf die
	// Systemseite gehört (siehe Hinweise).
	hinweise []string
}

// NewPostgres verbindet sich mit der Datenbank aus url, wartet nötigenfalls,
// bis sie bereit ist, und legt das Schema der Kontroll-Datenbank an.
func NewPostgres(ctx context.Context, url string) (*PostgresSpeicher, error) {
	basis, err := pgx.ParseConfig(url)
	if err != nil {
		// pgx ersetzt das Passwort in der Fehlermeldung durch "xxxxx".
		return nil, fmt.Errorf("DATABASE_URL ist ungültig: %w", err)
	}

	admin := stdlib.OpenDB(*basis)
	admin.SetMaxOpenConns(1)
	admin.SetConnMaxIdleTime(pgLeerlauf)

	p := &PostgresSpeicher{basis: basis, admin: admin, controlSchema: ControlSchema}

	if err := p.warteAufDatenbank(ctx); err != nil {
		admin.Close()
		return nil, err
	}

	p.pruefeSortierung(ctx)

	if err := p.legeSchemaAn(ctx, p.controlSchema); err != nil {
		admin.Close()
		return nil, err
	}
	return p, nil
}

// warteAufDatenbank versucht mehrmals, die Datenbank zu erreichen.
//
// Mit SQLite ist die Datei immer da. PostgreSQL läuft dagegen als eigener
// Dienst und ist nach einem gemeinsamen Start (docker compose up) oft ein paar
// Sekunden später bereit als Munin. Ohne Wiederholung bräche Munin dann sofort
// ab.
func (p *PostgresSpeicher) warteAufDatenbank(ctx context.Context) error {
	var letzterFehler error
	for versuch := 1; versuch <= pgStartVersuche; versuch++ {
		pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
		letzterFehler = p.admin.PingContext(pingCtx)
		cancel()
		if letzterFehler == nil {
			return nil
		}

		if versuch < pgStartVersuche {
			log.Printf("PostgreSQL ist noch nicht erreichbar (Versuch %d von %d): %v", versuch, pgStartVersuche, letzterFehler)
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(pgStartPause):
			}
		}
	}
	return fmt.Errorf("PostgreSQL ist nicht erreichbar (%s): %w", p.Beschreibung(), letzterFehler)
}

// pruefeSortierung warnt, wenn die Datenbank nicht mit der Sortierung "C"
// angelegt wurde.
//
// SQLite sortiert Text Byte für Byte. PostgreSQL sortiert nach der
// Spracheinstellung der Datenbank, und bei en_US/de_DE werden Bindestriche
// zunächst übergangen — "a-c" stünde dann vor "ab". Listen wären anders
// sortiert als unter SQLite. Kein Grund, den Start zu verweigern, aber einer
// für einen deutlichen Hinweis.
func (p *PostgresSpeicher) pruefeSortierung(ctx context.Context) {
	var sortierung string
	err := p.admin.QueryRowContext(ctx,
		`SELECT datcollate FROM pg_database WHERE datname = current_database()`,
	).Scan(&sortierung)
	if err != nil {
		log.Printf("Hinweis: Sortierung der Datenbank konnte nicht geprüft werden: %v", err)
		return
	}
	if sortierung != "C" && sortierung != "POSIX" {
		hinweis := fmt.Sprintf("Die Datenbank sortiert nach %q statt \"C\". Listen sind dann anders sortiert als unter SQLite. "+
			"Abhilfe: Datenbank mit --locale=C anlegen (siehe AUSBAUSTUFEN.md 4.10).", sortierung)
		log.Printf("Warnung: %s", hinweis)
		p.hinweise = append(p.hinweise, hinweis)
	}
}

// Hinweise nennt, was beim Start auffiel, etwa eine falsche Sortierung. Die
// Liste entsteht einmal in NewPostgres: Die Sortierung einer Datenbank lässt
// sich nachträglich nicht ändern, ein erneutes Nachfragen brächte nichts.
func (p *PostgresSpeicher) Hinweise() []string { return p.hinweise }

// OpenControl öffnet die Kontroll-Datenbank im Schema munin_control.
//
// Der ControlStore bekommt den Speicher als Haken mit: Er prüft vor jedem neuen
// Workspace die Verbindungen und legt danach dessen Schema samt lesbarem Namen
// an (siehe vorDemAnlegen, nachDemBenennen).
func (p *PostgresSpeicher) OpenControl() (*ControlStore, error) {
	lesen, schreiben, err := p.oeffnePools(p.controlSchema, controlMigrationsFS, "migrations_control")
	if err != nil {
		return nil, err
	}
	return &ControlStore{lesen: lesen, schreiben: schreiben, haken: p}, nil
}

// OpenWorkspace öffnet das Schema eines Workspace und legt es an, falls es
// noch fehlt.
func (p *PostgresSpeicher) OpenWorkspace(workspaceID string) (*Store, error) {
	schema, err := WorkspaceSchema(workspaceID)
	if err != nil {
		return nil, err
	}
	if err := p.legeSchemaAn(context.Background(), schema); err != nil {
		return nil, err
	}

	lesen, schreiben, err := p.oeffnePools(schema, migrationsFS, "migrations")
	if err != nil {
		return nil, err
	}
	return &Store{lesen: lesen, schreiben: schreiben}, nil
}

// DeleteWorkspace löscht das Schema eines Workspace samt aller Tabellen und
// Daten darin.
func (p *PostgresSpeicher) DeleteWorkspace(workspaceID string) error {
	schema, err := WorkspaceSchema(workspaceID)
	if err != nil {
		return err
	}
	if _, err := p.admin.ExecContext(context.Background(),
		`DROP SCHEMA IF EXISTS `+pgx.Identifier{schema}.Sanitize()+` CASCADE`,
	); err != nil {
		return fmt.Errorf("Schema %s konnte nicht gelöscht werden: %w", schema, err)
	}
	return nil
}

// Kapazitaet liest max_connections und nennt die Größen, mit denen
// PruefeKapazitaet rechnet.
func (p *PostgresSpeicher) Kapazitaet(ctx context.Context) (Kapazitaet, error) {
	erlaubt, err := p.maxConnections(ctx)
	if err != nil {
		return Kapazitaet{}, err
	}
	return Kapazitaet{
		Begrenzt: true, Erlaubt: erlaubt, Reserve: pgReserve,
		JeWorkspace: pgVerbindungenJeWorkspace, Fest: pgFesteVerbindungen,
	}, nil
}

// maxConnections liest max_connections des Servers.
func (p *PostgresSpeicher) maxConnections(ctx context.Context) (int, error) {
	var text string
	if err := p.admin.QueryRowContext(ctx, `SHOW max_connections`).Scan(&text); err != nil {
		return 0, fmt.Errorf("max_connections konnte nicht gelesen werden: %w", err)
	}
	erlaubt, err := strconv.Atoi(text)
	if err != nil {
		return 0, fmt.Errorf("max_connections hat einen unerwarteten Wert %q", text)
	}
	return erlaubt, nil
}

// PruefeKapazitaet rechnet nach, ob anzahl Workspaces mit ihren Pools unter
// max_connections bleiben (abzüglich Reserve).
//
// max_connections wird jedes Mal frisch gelesen: Wer es nach einer Meldung
// erhöht und PostgreSQL neu startet, soll Munin dafür nicht auch neu starten
// müssen.
func (p *PostgresSpeicher) PruefeKapazitaet(ctx context.Context, anzahl int) error {
	erlaubt, err := p.maxConnections(ctx)
	if err != nil {
		return err
	}

	bedarf := anzahl*pgVerbindungenJeWorkspace + pgFesteVerbindungen
	if bedarf > erlaubt-pgReserve {
		return fmt.Errorf("zu wenige PostgreSQL-Verbindungen: %d Workspaces brauchen bis zu %d Verbindungen "+
			"(%d je Workspace + %d fest), max_connections ist %d (davon %d Reserve). "+
			"max_connections auf mindestens %d erhöhen",
			anzahl, bedarf, pgVerbindungenJeWorkspace, pgFesteVerbindungen, erlaubt, pgReserve, bedarf+pgReserve)
	}
	return nil
}

// Ort nennt das Schema des Workspace, z.B. "munin_ws_mue9bh3x_1c0e1120".
// Eine ID, die keinen gültigen Schemanamen ergibt, kann es hier nicht geben
// (OpenWorkspace hätte sie abgelehnt); falls doch, steht der Grund da.
func (p *PostgresSpeicher) Ort(workspaceID string) string {
	schema, err := WorkspaceSchema(workspaceID)
	if err != nil {
		return err.Error()
	}
	return schema
}

// Beschreibung nennt Server und Datenbank, nie das Passwort.
func (p *PostgresSpeicher) Beschreibung() string {
	return fmt.Sprintf("PostgreSQL — %s:%d/%s (Schemas %s, %s…)",
		p.basis.Host, p.basis.Port, p.basis.Database, ControlSchema, workspaceSchemaPraefix)
}

// Close schließt die Verwaltungsverbindung.
func (p *PostgresSpeicher) Close() error { return p.admin.Close() }

// ── Haken für den ControlStore ───────────────────────────────────────────────

func (p *PostgresSpeicher) vorDemAnlegen(ctx context.Context, anzahl int) error {
	return p.PruefeKapazitaet(ctx, anzahl)
}

// nachDemBenennen legt das Schema des Workspace an (falls noch nicht da) und
// schreibt den lesbaren Namen als Kommentar daran. In psql zeigt \dn+ ihn
// neben dem Schemanamen.
func (p *PostgresSpeicher) nachDemBenennen(ctx context.Context, workspaceID, name string) error {
	schema, err := WorkspaceSchema(workspaceID)
	if err != nil {
		return err
	}
	if err := p.legeSchemaAn(ctx, schema); err != nil {
		return err
	}

	// COMMENT ON nimmt keine Platzhalter ($1) an, der Name muss also als
	// Text in die Anweisung. textLiteral sorgt dafür, dass er dort nichts
	// anderes sein kann als Text.
	if _, err := p.admin.ExecContext(ctx,
		`COMMENT ON SCHEMA `+pgx.Identifier{schema}.Sanitize()+` IS `+textLiteral(name),
	); err != nil {
		return fmt.Errorf("Name konnte nicht am Schema %s vermerkt werden: %w", schema, err)
	}
	return nil
}

// ── Hilfsfunktionen ──────────────────────────────────────────────────────────

func (p *PostgresSpeicher) legeSchemaAn(ctx context.Context, schema string) error {
	if _, err := p.admin.ExecContext(ctx,
		`CREATE SCHEMA IF NOT EXISTS `+pgx.Identifier{schema}.Sanitize(),
	); err != nil {
		return fmt.Errorf("Schema %s konnte nicht angelegt werden: %w", schema, err)
	}
	return nil
}

// oeffnePools öffnet Lese- und Schreibpool für ein Schema und bringt dessen
// Tabellen auf den aktuellen Stand — das Gegenstück zu openSQLite.
func (p *PostgresSpeicher) oeffnePools(schema string, fsys embed.FS, migrationsDir string) (lesen, schreiben pool, err error) {
	d := postgresDialekt{}

	// Jede Verbindung dieser Pools startet mit search_path=<schema>. Kopie,
	// damit die Grundeinstellung für die anderen Schemas unverändert bleibt.
	cfg := p.basis.Copy()
	cfg.RuntimeParams["search_path"] = schema

	oeffne := func(maxVerbindungen int) *sql.DB {
		db := stdlib.OpenDB(*cfg)
		db.SetMaxOpenConns(maxVerbindungen)
		db.SetMaxIdleConns(maxVerbindungen)
		db.SetConnMaxIdleTime(pgLeerlauf)
		return db
	}

	schreibDB := oeffne(pgSchreibverbindungen)
	schreiben = pool{db: schreibDB, d: d}
	if err := runMigrations(schreiben, fsys, migrationsDir); err != nil {
		schreibDB.Close()
		return pool{}, pool{}, fmt.Errorf("Schema %s: %w", schema, err)
	}

	lesen = pool{db: oeffne(pgLeseverbindungen), d: d}
	return lesen, schreiben, nil
}

// textLiteral schreibt s als PostgreSQL-Textkonstante, z.B. E'Vertrieb & Co'.
//
// Die Form E'…' wertet Backslashes immer gleich aus, unabhängig von der
// Einstellung standard_conforming_strings. Darin werden Backslash und
// Hochkomma verdoppelt; ein Nullbyte kann PostgreSQL in Text ohnehin nicht
// speichern und wird entfernt.
func textLiteral(s string) string {
	s = strings.ReplaceAll(s, "\x00", "")
	s = strings.ReplaceAll(s, `\`, `\\`)
	s = strings.ReplaceAll(s, `'`, `''`)
	return "E'" + s + "'"
}