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 }