syncova-backup/packages/agentregistry/store_postgres.go

package agentregistry

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

// PostgresStore legt die Agent-Daten in PostgreSQL ab.
type PostgresStore struct {
	// connectionPool ist der Datenbankpool der Control Plane.
	connectionPool *pgxpool.Pool
}

// NewPostgresStore erzeugt die Datenzugriffsschicht.
func NewPostgresStore(connectionPool *pgxpool.Pool) *PostgresStore {
	return &PostgresStore{connectionPool: connectionPool}
}

// CreateEnrollmentToken legt ein Aufnahme-Token an.
func (store *PostgresStore) CreateEnrollmentToken(createContext context.Context, tokenHash string, agentName string, expiresAt time.Time, createdBy *uuid.UUID) (uuid.UUID, error) {
	const insertStatement = `
		INSERT INTO agent_enrollment_tokens (token_hash, agent_name, expires_at, created_by)
		VALUES ($1, $2, $3, $4)
		RETURNING id`

	var tokenID uuid.UUID
	if scanError := store.connectionPool.QueryRow(createContext, insertStatement,
		tokenHash, agentName, expiresAt, createdBy).Scan(&tokenID); scanError != nil {
		return uuid.Nil, fmt.Errorf("das aufnahme-token konnte nicht angelegt werden: %w", scanError)
	}

	return tokenID, nil
}

// PeekEnrollmentToken liest den vorgesehenen Namen, ohne einzulösen.
//
// Abgelaufene und bereits eingelöste Tokens werden bereits in der Abfrage
// ausgeschlossen: so kann ein vergessener Zustandsvergleich im Code nicht zu
// einer unberechtigten Aufnahme führen.
func (store *PostgresStore) PeekEnrollmentToken(peekContext context.Context, tokenHash string) (string, error) {
	const selectStatement = `
		SELECT agent_name
		FROM agent_enrollment_tokens
		WHERE token_hash = $1 AND used_at IS NULL AND expires_at > now()`

	var agentName string
	scanError := store.connectionPool.QueryRow(peekContext, selectStatement, tokenHash).Scan(&agentName)

	if errors.Is(scanError, pgx.ErrNoRows) {
		return "", ErrEnrollmentTokenInvalid
	}

	if scanError != nil {
		return "", fmt.Errorf("das aufnahme-token konnte nicht geprüft werden: %w", scanError)
	}

	return agentName, nil
}

// ConsumeEnrollmentToken löst ein Aufnahme-Token ein.
//
// Die Bedingung auf used_at macht das Einlösen unteilbar: zwei gleichzeitige
// Registrierungen können nicht beide erfolgreich sein.
func (store *PostgresStore) ConsumeEnrollmentToken(consumeContext context.Context, tokenHash string, agentID uuid.UUID) (string, error) {
	const updateStatement = `
		UPDATE agent_enrollment_tokens
		SET used_at = now(), used_by_agent_id = $2
		WHERE token_hash = $1 AND used_at IS NULL AND expires_at > now()
		RETURNING agent_name`

	var agentName string
	scanError := store.connectionPool.QueryRow(consumeContext, updateStatement, tokenHash, agentID).Scan(&agentName)

	if errors.Is(scanError, pgx.ErrNoRows) {
		return "", ErrEnrollmentTokenInvalid
	}

	if scanError != nil {
		return "", fmt.Errorf("das aufnahme-token konnte nicht eingelöst werden: %w", scanError)
	}

	return agentName, nil
}

// CreateAgent legt einen Agent an.
func (store *PostgresStore) CreateAgent(createContext context.Context, agentToCreate Agent) (uuid.UUID, error) {
	const insertStatement = `
		INSERT INTO agents (name, hostname, platform, architecture, version, status, registered_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7)
		RETURNING id`

	var agentID uuid.UUID
	if scanError := store.connectionPool.QueryRow(createContext, insertStatement,
		agentToCreate.Name, nullIfEmpty(agentToCreate.Hostname), string(agentToCreate.Platform),
		nullIfEmpty(agentToCreate.Architecture), nullIfEmpty(agentToCreate.Version),
		string(agentToCreate.Status), agentToCreate.RegisteredAt).Scan(&agentID); scanError != nil {
		return uuid.Nil, fmt.Errorf("der agent konnte nicht angelegt werden: %w", scanError)
	}

	return agentID, nil
}

// CreateAgentToken legt ein Betriebstoken an.
func (store *PostgresStore) CreateAgentToken(createContext context.Context, agentID uuid.UUID, tokenHash string) (uuid.UUID, error) {
	const insertStatement = `
		INSERT INTO agent_tokens (agent_id, token_hash)
		VALUES ($1, $2)
		RETURNING id`

	var tokenID uuid.UUID
	if scanError := store.connectionPool.QueryRow(createContext, insertStatement,
		agentID, tokenHash).Scan(&tokenID); scanError != nil {
		return uuid.Nil, fmt.Errorf("das agent-token konnte nicht angelegt werden: %w", scanError)
	}

	return tokenID, nil
}

// FindAgentByTokenHash sucht einen Agent anhand seines Betriebstokens.
func (store *PostgresStore) FindAgentByTokenHash(queryContext context.Context, tokenHash string) (Agent, error) {
	const selectStatement = `
		SELECT a.id, a.name, coalesce(a.hostname, ''), a.platform, coalesce(a.architecture, ''),
		       coalesce(a.version, ''), a.status, a.last_heartbeat_at,
		       coalesce(host(a.last_ip_address), ''), a.registered_at
		FROM agent_tokens t
		JOIN agents a ON a.id = t.agent_id
		WHERE t.token_hash = $1
		  AND t.revoked_at IS NULL
		  AND (t.expires_at IS NULL OR t.expires_at > now())
		  AND a.revoked_at IS NULL`

	foundAgent, scanError := scanAgentRow(store.connectionPool.QueryRow(queryContext, selectStatement, tokenHash))

	if errors.Is(scanError, pgx.ErrNoRows) {
		return Agent{}, ErrAgentTokenInvalid
	}

	if scanError != nil {
		return Agent{}, fmt.Errorf("das agent-token konnte nicht geprüft werden: %w", scanError)
	}

	// Die Verwendung wird vermerkt, damit sich ein stillgelegtes Token erkennen lässt.
	if _, updateError := store.connectionPool.Exec(queryContext,
		"UPDATE agent_tokens SET last_used_at = now() WHERE token_hash = $1", tokenHash); updateError != nil {
		// Ein fehlgeschlagener Vermerk darf die Anmeldung nicht verhindern.
		_ = updateError
	}

	return foundAgent, nil
}

// FindAgentByID sucht einen Agent anhand seiner Kennung.
func (store *PostgresStore) FindAgentByID(queryContext context.Context, agentID uuid.UUID) (Agent, error) {
	const selectStatement = `
		SELECT id, name, coalesce(hostname, ''), platform, coalesce(architecture, ''),
		       coalesce(version, ''), status, last_heartbeat_at,
		       coalesce(host(last_ip_address), ''), registered_at
		FROM agents
		WHERE id = $1`

	foundAgent, scanError := scanAgentRow(store.connectionPool.QueryRow(queryContext, selectStatement, agentID))

	if errors.Is(scanError, pgx.ErrNoRows) {
		return Agent{}, ErrAgentNotFound
	}

	if scanError != nil {
		return Agent{}, fmt.Errorf("der agent konnte nicht gelesen werden: %w", scanError)
	}

	return foundAgent, nil
}

// ListAgents liefert eine Seite von Agents.
func (store *PostgresStore) ListAgents(queryContext context.Context, agentFilter AgentFilter) ([]Agent, int64, error) {
	whereClause := " WHERE 1=1"
	queryArguments := make([]any, 0, 2)

	if agentFilter.Status != "" {
		queryArguments = append(queryArguments, agentFilter.Status)
		whereClause += fmt.Sprintf(" AND status = $%d", len(queryArguments))
	}

	if agentFilter.Platform != "" {
		queryArguments = append(queryArguments, agentFilter.Platform)
		whereClause += fmt.Sprintf(" AND platform = $%d", len(queryArguments))
	}

	var totalCount int64
	if scanError := store.connectionPool.QueryRow(queryContext,
		"SELECT count(*) FROM agents"+whereClause, queryArguments...).Scan(&totalCount); scanError != nil {
		return nil, 0, fmt.Errorf("die anzahl der agents konnte nicht ermittelt werden: %w", scanError)
	}

	listStatement := `
		SELECT id, name, coalesce(hostname, ''), platform, coalesce(architecture, ''),
		       coalesce(version, ''), status, last_heartbeat_at,
		       coalesce(host(last_ip_address), ''), registered_at
		FROM agents` + whereClause + fmt.Sprintf(
		" ORDER BY name LIMIT $%d OFFSET $%d", len(queryArguments)+1, len(queryArguments)+2)

	queryArguments = append(queryArguments, agentFilter.PageSize, (agentFilter.Page-1)*agentFilter.PageSize)

	agentRows, queryError := store.connectionPool.Query(queryContext, listStatement, queryArguments...)
	if queryError != nil {
		return nil, 0, fmt.Errorf("die agents konnten nicht gelesen werden: %w", queryError)
	}
	defer agentRows.Close()

	loadedAgents := make([]Agent, 0, agentFilter.PageSize)
	for agentRows.Next() {
		loadedAgent, scanError := scanAgentRow(agentRows)
		if scanError != nil {
			return nil, 0, fmt.Errorf("ein agent konnte nicht gelesen werden: %w", scanError)
		}

		loadedAgents = append(loadedAgents, loadedAgent)
	}

	if rowsError := agentRows.Err(); rowsError != nil {
		return nil, 0, fmt.Errorf("die agents konnten nicht vollständig gelesen werden: %w", rowsError)
	}

	return loadedAgents, totalCount, nil
}

// RecordHeartbeat vermerkt eine Lebendmeldung.
func (store *PostgresStore) RecordHeartbeat(updateContext context.Context, agentID uuid.UUID, agentVersion string, ipAddress string) error {
	const updateStatement = `
		UPDATE agents
		SET last_heartbeat_at = now(),
		    version = coalesce(nullif($2, ''), version),
		    last_ip_address = coalesce($3::inet, last_ip_address),
		    updated_at = now()
		WHERE id = $1 AND revoked_at IS NULL`

	commandTag, updateError := store.connectionPool.Exec(updateContext, updateStatement,
		agentID, agentVersion, nullIfEmpty(ipAddress))

	if updateError != nil {
		return fmt.Errorf("die lebendmeldung konnte nicht vermerkt werden: %w", updateError)
	}

	if commandTag.RowsAffected() == 0 {
		return ErrAgentNotFound
	}

	return nil
}

// RevokeAgent sperrt einen Agent und alle seine Tokens.
func (store *PostgresStore) RevokeAgent(updateContext context.Context, agentID uuid.UUID) error {
	databaseTransaction, transactionError := store.connectionPool.Begin(updateContext)
	if transactionError != nil {
		return fmt.Errorf("die transaktion konnte nicht begonnen werden: %w", transactionError)
	}
	defer func() { _ = databaseTransaction.Rollback(updateContext) }()

	commandTag, updateError := databaseTransaction.Exec(updateContext, `
		UPDATE agents SET status = 'revoked', revoked_at = now(), updated_at = now()
		WHERE id = $1 AND revoked_at IS NULL`, agentID)

	if updateError != nil {
		return fmt.Errorf("der agent konnte nicht gesperrt werden: %w", updateError)
	}

	if commandTag.RowsAffected() == 0 {
		return ErrAgentNotFound
	}

	// Ohne Widerruf der Tokens könnte sich der gesperrte Agent weiter melden.
	if _, tokenError := databaseTransaction.Exec(updateContext,
		"UPDATE agent_tokens SET revoked_at = now() WHERE agent_id = $1 AND revoked_at IS NULL", agentID); tokenError != nil {
		return fmt.Errorf("die tokens konnten nicht widerrufen werden: %w", tokenError)
	}

	if commitError := databaseTransaction.Commit(updateContext); commitError != nil {
		return fmt.Errorf("die sperre konnte nicht gespeichert werden: %w", commitError)
	}

	return nil
}

// RotateAgentToken widerruft alle Tokens eines Agents und legt ein neues an.
func (store *PostgresStore) RotateAgentToken(updateContext context.Context, agentID uuid.UUID, newTokenHash string) error {
	databaseTransaction, transactionError := store.connectionPool.Begin(updateContext)
	if transactionError != nil {
		return fmt.Errorf("die transaktion konnte nicht begonnen werden: %w", transactionError)
	}
	defer func() { _ = databaseTransaction.Rollback(updateContext) }()

	var newTokenID uuid.UUID
	if scanError := databaseTransaction.QueryRow(updateContext,
		"INSERT INTO agent_tokens (agent_id, token_hash) VALUES ($1, $2) RETURNING id",
		agentID, newTokenHash).Scan(&newTokenID); scanError != nil {
		return fmt.Errorf("das neue token konnte nicht angelegt werden: %w", scanError)
	}

	// Der Verweis auf das Nachfolgetoken macht die Rotation nachvollziehbar.
	if _, revokeError := databaseTransaction.Exec(updateContext, `
		UPDATE agent_tokens
		SET revoked_at = now(), replaced_by_token_id = $2
		WHERE agent_id = $1 AND revoked_at IS NULL AND id <> $2`, agentID, newTokenID); revokeError != nil {
		return fmt.Errorf("die bisherigen tokens konnten nicht widerrufen werden: %w", revokeError)
	}

	if commitError := databaseTransaction.Commit(updateContext); commitError != nil {
		return fmt.Errorf("der tokenwechsel konnte nicht gespeichert werden: %w", commitError)
	}

	return nil
}

// rowScanner deckt sowohl pgx.Row als auch pgx.Rows ab.
type rowScanner interface {
	Scan(destinations ...any) error
}

// scanAgentRow liest eine Agent-Zeile.
func scanAgentRow(scannableRow rowScanner) (Agent, error) {
	var scannedAgent Agent
	var platformValue string
	var statusValue string

	scanError := scannableRow.Scan(
		&scannedAgent.ID, &scannedAgent.Name, &scannedAgent.Hostname, &platformValue,
		&scannedAgent.Architecture, &scannedAgent.Version, &statusValue,
		&scannedAgent.LastHeartbeatAt, &scannedAgent.LastIPAddress, &scannedAgent.RegisteredAt,
	)

	if scanError != nil {
		return Agent{}, scanError
	}

	scannedAgent.Platform = Platform(platformValue)
	scannedAgent.Status = AgentStatus(statusValue)

	return scannedAgent, nil
}

// nullIfEmpty wandelt eine leere Zeichenkette in NULL.
func nullIfEmpty(textValue string) *string {
	if textValue == "" {
		return nil
	}

	return &textValue
}