feat: add proxmox reconciliation job

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Proxmox-Reconciliation-Job fuer aktive Cluster angelegt; Drift wird report-only als Audit-Event geloggt.
 - VM-Power-Endpunkte fuer start/stop/reboot mit Proxmox-UPID, Queue-Enqueue und Audit-Write angelegt.
 - VM-Read-Endpunkte fuer Projekt-Listen und VM-Details mit Membership-gefilterten SQL-Queries angelegt.
 - Cluster-, Krypto- und Proxmox-Client-Bausteine nach `platform/` verschoben und im Worker fuer echte UPID-Polls verdrahtet.

README.md

@@ -40,7 +40,7 @@ Der Proxmox-Client nutzt HTTPS mit normaler Zertifikatsvalidierung plus SHA-256-
 
 Interne Cluster-Admin-Routen unter `/internal/clusters` sind nicht tenant-gebunden. Sie erfordern den Header `X-ProxUI-Operator-Token` mit `OPERATOR_TOKEN` und die isolierte Rolle `operator`; Responses enthalten nie `token_secret`.
 
-Der Worker nutzt `asynq` gegen `REDIS_ADDR`, pingt beim Start dieselbe Supabase-DB ueber `DATABASE_URL`, entschluesselt Cluster-Credentials mit `MASTER_KEY_BASE64` und nutzt den gemeinsamen Proxmox-Client aus `platform/proxmox` fuer UPID-Polls.
+Der Worker nutzt `asynq` gegen `REDIS_ADDR`, pingt beim Start dieselbe Supabase-DB ueber `DATABASE_URL`, entschluesselt Cluster-Credentials mit `MASTER_KEY_BASE64` und nutzt den gemeinsamen Proxmox-Client aus `platform/proxmox` fuer UPID-Polls und Reconciliation.
 
 Lokale Dienste:
 
@@ -66,6 +66,7 @@ Worker-Jobs:
 
 - `proxui.dummy`: Dummy-Job mit Payload `{"message":"..."}` fuer Smoke-Tests der asynq-Verarbeitung
 - `proxmox.task.poll`: Pollt einen Proxmox-UPID ueber den gespeicherten Cluster, setzt VM-Status auf Erfolgs- oder Fehlerstatus und schreibt Audit-Metadaten.
+- `proxmox.reconcile.all`: Laeuft alle 5 Minuten, prueft aktive Cluster gegen bekannte VMs und schreibt Drift als Audit-Event `vm.reconcile.drift`.
 
 Backend-Endpunkte:
 

TODO.md

@@ -151,6 +151,12 @@ Arbeitsliste auf Basis von `proxmox-console-entwicklungsplan.md`. Die Entwurfsda
   - [x] `vm.power` wird anhand der Membership-Rolle geprueft
   - [x] Proxmox-Call liefert UPID und queued `proxmox.task.poll`
   - [x] Audit-Log-Eintrag fuer ausgeloeste Power-Aktion wird geschrieben
+- [x] E5-T03: Reconciliation-Job
+  - [x] Periodischen asynq-Scheduler fuer `proxmox.reconcile.all` angelegt
+  - [x] Aktive Cluster werden aus der DB geladen
+  - [x] Bekannte VMs je Cluster werden gegen Proxmox `status/current` geprueft
+  - [x] Drift wird als Audit-Event `vm.reconcile.drift` geloggt
+  - [x] Kein Auto-Fix: DB-Status wird im Reconciliation-Lauf nicht ueberschrieben
 
 ## MVP-Backlog
 
@@ -159,7 +165,7 @@ Arbeitsliste auf Basis von `proxmox-console-entwicklungsplan.md`. Die Entwurfsda
 - [x] E3: Envelope-Encryption, Cluster-Repository, Proxmox-Client und interne Cluster-Verwaltung
 - [ ] E4: RBAC, Membership-Resolver, Policy-Funktion und Autorisierungs-Middleware
 - [ ] E6: Worker-Grundgeruest und UPID-Polling
-- [ ] E5: VM-Liste, Detail, Power-Aktionen und Reconciliation
+- [x] E5: VM-Liste, Detail, Power-Aktionen und Reconciliation
 - [ ] E7: SSH-Keys, Templates und Provisioning
 - [ ] E8: Konsolen-Tickets und Websocket-Proxy
 - [ ] E9: Audit-Writer und Audit-Anzeige
@@ -213,3 +219,4 @@ Arbeitsliste auf Basis von `proxmox-console-entwicklungsplan.md`. Die Entwurfsda
 - 2026-06-11: Shared Cluster-/Krypto-/Proxmox-Packages nach `platform/` verschoben und Worker-UPID-Polling an echte Cluster-Aufloesung angeschlossen.
 - 2026-06-11: VM-Read-Endpunkte fuer Projekt-Listen und Details mit Membership-gefilterten Queries angelegt.
 - 2026-06-11: VM-Power-Endpunkte fuer start/stop/reboot mit Proxmox-UPID, Queue-Enqueue und Audit-Write angelegt.
+- 2026-06-11: Reconciliation-Job fuer aktive Proxmox-Cluster angelegt; Drift wird report-only auditiert.

platform/jobs/proxmox_task.go

@@ -1,6 +1,7 @@
 package jobs
 
 const TypeProxmoxTaskPoll = "proxmox.task.poll"
+const TypeProxmoxReconcileAll = "proxmox.reconcile.all"
 
 type ProxmoxTaskPollPayload struct {
 	ClusterID     string `json:"cluster_id"`

platform/proxmox/client.go

@@ -131,6 +131,10 @@ type TaskStatus struct {
 	ExitStatus string
 }
 
+type VMStatus struct {
+	Status string
+}
+
 func (c *Client) GetTaskStatus(ctx context.Context, node string, upid string) (TaskStatus, error) {
 	response, err := c.Get(ctx, fmt.Sprintf(
 		"/nodes/%s/tasks/%s/status",
@@ -193,6 +197,37 @@ func (c *Client) PowerVM(ctx context.Context, node string, vmid int, action stri
 	return body.Data, nil
 }
 
+func (c *Client) GetVMStatus(ctx context.Context, node string, vmid int) (VMStatus, error) {
+	response, err := c.Get(ctx, fmt.Sprintf(
+		"/nodes/%s/qemu/%d/status/current",
+		url.PathEscape(node),
+		vmid,
+	))
+	if err != nil {
+		return VMStatus{}, err
+	}
+	defer response.Body.Close()
+
+	if response.StatusCode >= http.StatusBadRequest {
+		_, _ = io.Copy(io.Discard, response.Body)
+		return VMStatus{}, fmt.Errorf("proxmox returned %s", response.Status)
+	}
+
+	var body struct {
+		Data struct {
+			Status string `json:"status"`
+		} `json:"data"`
+	}
+	if err := json.NewDecoder(response.Body).Decode(&body); err != nil {
+		return VMStatus{}, err
+	}
+	if body.Data.Status == "" {
+		return VMStatus{}, fmt.Errorf("proxmox response missing VM status")
+	}
+
+	return VMStatus{Status: body.Data.Status}, nil
+}
+
 func (c *Client) Do(ctx context.Context, method string, path string, body []byte) (*http.Response, error) {
 	var lastErr error
 	attempts := c.retries + 1

platform/proxmox/client_test.go

@@ -130,6 +130,25 @@ func TestClientPowersVM(t *testing.T) {
 	}
 }
 
+func TestClientGetsVMStatus(t *testing.T) {
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/api2/json/nodes/pve/qemu/100/status/current" {
+			t.Fatalf("path = %q", r.URL.Path)
+		}
+		_, _ = w.Write([]byte(`{"data":{"status":"running"}}`))
+	}))
+	defer server.Close()
+
+	client := newTestClient(t, server, fingerprintForServer(server))
+	status, err := client.GetVMStatus(context.Background(), "pve", 100)
+	if err != nil {
+		t.Fatalf("GetVMStatus() error = %v", err)
+	}
+	if status.Status != "running" {
+		t.Fatalf("status = %q, want running", status.Status)
+	}
+}
+
 func TestNewClientRejectsInvalidFingerprint(t *testing.T) {
 	_, err := NewClient(cluster.Cluster{
 		APIEndpoint:    "https://pve.example.test:8006/api2/json",

worker/cmd/worker/main.go

@@ -50,9 +50,14 @@ func main() {
 	server := asynq.NewServer(redis, asynq.Config{
 		Concurrency: cfg.WorkerConcurrency,
 	})
+	scheduler := asynq.NewScheduler(redis, &asynq.SchedulerOpts{})
 	mux := asynq.NewServeMux()
 	clusterRepository := cluster.NewRepository(db, tokenCipher)
 	registerHandlers(mux, db, clusterRepository, logger)
+	if _, err := scheduler.Register("@every 5m", tasks.NewProxmoxReconcileAllTask()); err != nil {
+		logger.Error("failed to register reconciliation schedule", "error", err)
+		os.Exit(1)
+	}
 
 	go func() {
 		logger.Info("worker started", "concurrency", cfg.WorkerConcurrency, "redis_addr", cfg.RedisAddr)
@@ -61,8 +66,16 @@ func main() {
 			stop()
 		}
 	}()
+	go func() {
+		logger.Info("worker scheduler started", "reconcile_interval", "5m")
+		if err := scheduler.Run(); err != nil {
+			logger.Error("worker scheduler failed", "error", err)
+			stop()
+		}
+	}()
 
 	<-ctx.Done()
+	scheduler.Shutdown()
 	server.Shutdown()
 	logger.Info("worker stopped")
 }
@@ -78,6 +91,17 @@ func registerHandlers(mux *asynq.ServeMux, db *sql.DB, clusters cluster.Reposito
 			logger,
 		),
 	)
+	reconcileStore := tasks.NewSQLReconcileStore(db)
+	mux.Handle(
+		tasks.TypeProxmoxReconcileAll,
+		tasks.NewProxmoxReconcileHandler(
+			reconcileStore,
+			reconcileStore,
+			tasks.NewDefaultClusterReconcileClientResolver(clusters),
+			tasks.NewSQLAuditWriter(db),
+			logger,
+		),
+	)
 }
 
 func openDatabase(databaseURL string) (*sql.DB, error) {

worker/internal/tasks/proxmox_reconcile.go

@@ -0,0 +1,211 @@
+package tasks
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+
+	"forgejo.digital-droplets.de/philschlo/proxui/platform/jobs"
+	"github.com/hibiken/asynq"
+)
+
+const TypeProxmoxReconcileAll = jobs.TypeProxmoxReconcileAll
+
+type ReconcileClusterStore interface {
+	ListActiveClusterIDs(ctx context.Context) ([]string, error)
+}
+
+type ReconcileVMStore interface {
+	ListClusterVMs(ctx context.Context, clusterID string) ([]ReconcileVM, error)
+}
+
+type ReconcileVM struct {
+	ID          string
+	TenantID    string
+	Node        string
+	ProxmoxVMID int
+	Status      string
+}
+
+type ProxmoxReconcileClient interface {
+	GetVMStatus(ctx context.Context, node string, vmid int) (ProxmoxVMStatus, error)
+}
+
+type ProxmoxVMStatus struct {
+	Status string
+}
+
+type ProxmoxReconcileClientResolver interface {
+	ResolveReconcileClient(ctx context.Context, clusterID string) (ProxmoxReconcileClient, error)
+}
+
+type ProxmoxReconcileHandler struct {
+	clusters    ReconcileClusterStore
+	vms         ReconcileVMStore
+	resolver    ProxmoxReconcileClientResolver
+	auditWriter AuditWriter
+	logger      *slog.Logger
+}
+
+func NewProxmoxReconcileHandler(
+	clusters ReconcileClusterStore,
+	vms ReconcileVMStore,
+	resolver ProxmoxReconcileClientResolver,
+	auditWriter AuditWriter,
+	logger *slog.Logger,
+) ProxmoxReconcileHandler {
+	return ProxmoxReconcileHandler{
+		clusters:    clusters,
+		vms:         vms,
+		resolver:    resolver,
+		auditWriter: auditWriter,
+		logger:      logger,
+	}
+}
+
+func NewProxmoxReconcileAllTask() *asynq.Task {
+	return asynq.NewTask(TypeProxmoxReconcileAll, nil)
+}
+
+func (h ProxmoxReconcileHandler) ProcessTask(ctx context.Context, _ *asynq.Task) error {
+	clusterIDs, err := h.clusters.ListActiveClusterIDs(ctx)
+	if err != nil {
+		return fmt.Errorf("list active clusters: %w", err)
+	}
+
+	for _, clusterID := range clusterIDs {
+		if err := h.reconcileCluster(ctx, clusterID); err != nil {
+			return err
+		}
+	}
+
+	h.logger.Info("completed proxmox reconciliation", "cluster_count", len(clusterIDs))
+	return nil
+}
+
+func (h ProxmoxReconcileHandler) reconcileCluster(ctx context.Context, clusterID string) error {
+	client, err := h.resolver.ResolveReconcileClient(ctx, clusterID)
+	if err != nil {
+		return fmt.Errorf("resolve proxmox reconcile client for cluster %s: %w", clusterID, err)
+	}
+
+	vms, err := h.vms.ListClusterVMs(ctx, clusterID)
+	if err != nil {
+		return fmt.Errorf("list cluster vms for cluster %s: %w", clusterID, err)
+	}
+
+	driftCount := 0
+	for _, vm := range vms {
+		proxmoxStatus, err := client.GetVMStatus(ctx, vm.Node, vm.ProxmoxVMID)
+		if err != nil {
+			return fmt.Errorf("get proxmox vm status for vm %s: %w", vm.ID, err)
+		}
+
+		currentStatus := normalizeProxmoxVMStatus(proxmoxStatus.Status)
+		if currentStatus == "" || currentStatus == vm.Status {
+			continue
+		}
+
+		driftCount++
+		if err := h.auditWriter.WriteAudit(ctx, AuditEvent{
+			TenantID:   vm.TenantID,
+			Action:     "vm.reconcile.drift",
+			TargetType: "vm",
+			TargetID:   vm.ID,
+			Metadata: map[string]any{
+				"cluster_id":     clusterID,
+				"node":           vm.Node,
+				"proxmox_vmid":   vm.ProxmoxVMID,
+				"db_status":      vm.Status,
+				"proxmox_status": currentStatus,
+			},
+		}); err != nil {
+			return fmt.Errorf("write reconcile drift audit for vm %s: %w", vm.ID, err)
+		}
+	}
+
+	h.logger.Info("reconciled proxmox cluster", "cluster_id", clusterID, "vm_count", len(vms), "drift_count", driftCount)
+	return nil
+}
+
+func normalizeProxmoxVMStatus(status string) string {
+	switch status {
+	case "running":
+		return "running"
+	case "stopped":
+		return "stopped"
+	case "paused":
+		return "suspended"
+	default:
+		return ""
+	}
+}
+
+type SQLReconcileStore struct {
+	db *sql.DB
+}
+
+func NewSQLReconcileStore(db *sql.DB) SQLReconcileStore {
+	return SQLReconcileStore{db: db}
+}
+
+func (s SQLReconcileStore) ListActiveClusterIDs(ctx context.Context) ([]string, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		select id::text
+		from public.clusters
+		where status = 'active'
+		order by created_at asc
+	`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var clusterIDs []string
+	for rows.Next() {
+		var clusterID string
+		if err := rows.Scan(&clusterID); err != nil {
+			return nil, err
+		}
+		clusterIDs = append(clusterIDs, clusterID)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return clusterIDs, nil
+}
+
+func (s SQLReconcileStore) ListClusterVMs(ctx context.Context, clusterID string) ([]ReconcileVM, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		select
+			v.id::text,
+			p.tenant_id::text,
+			v.node,
+			v.proxmox_vmid,
+			v.status::text
+		from public.vms v
+		join public.projects p
+			on p.id = v.project_id
+		where v.cluster_id = $1
+			and v.status not in ('deleting', 'deleted')
+		order by v.created_at asc
+	`, clusterID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var vms []ReconcileVM
+	for rows.Next() {
+		var vm ReconcileVM
+		if err := rows.Scan(&vm.ID, &vm.TenantID, &vm.Node, &vm.ProxmoxVMID, &vm.Status); err != nil {
+			return nil, err
+		}
+		vms = append(vms, vm)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return vms, nil
+}

worker/internal/tasks/proxmox_reconcile_resolver.go

@@ -0,0 +1,62 @@
+package tasks
+
+import (
+	"context"
+
+	"forgejo.digital-droplets.de/philschlo/proxui/platform/cluster"
+	"forgejo.digital-droplets.de/philschlo/proxui/platform/proxmox"
+)
+
+type ProxmoxReconcileClientFactory func(cluster cluster.Cluster) (PlatformProxmoxReconcileClient, error)
+
+type PlatformProxmoxReconcileClient interface {
+	GetVMStatus(ctx context.Context, node string, vmid int) (proxmox.VMStatus, error)
+}
+
+type ClusterReconcileClientResolver struct {
+	clusters ClusterRepository
+	factory  ProxmoxReconcileClientFactory
+}
+
+func NewClusterReconcileClientResolver(clusters ClusterRepository, factory ProxmoxReconcileClientFactory) ClusterReconcileClientResolver {
+	return ClusterReconcileClientResolver{
+		clusters: clusters,
+		factory:  factory,
+	}
+}
+
+func NewDefaultClusterReconcileClientResolver(clusters ClusterRepository) ClusterReconcileClientResolver {
+	return NewClusterReconcileClientResolver(clusters, func(cluster cluster.Cluster) (PlatformProxmoxReconcileClient, error) {
+		return proxmox.NewClient(cluster)
+	})
+}
+
+func (r ClusterReconcileClientResolver) ResolveReconcileClient(ctx context.Context, clusterID string) (ProxmoxReconcileClient, error) {
+	cluster, found, err := r.clusters.GetCluster(ctx, clusterID)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, ErrClusterNotFound
+	}
+
+	client, err := r.factory(cluster)
+	if err != nil {
+		return nil, err
+	}
+
+	return platformReconcileClientAdapter{client: client}, nil
+}
+
+type platformReconcileClientAdapter struct {
+	client PlatformProxmoxReconcileClient
+}
+
+func (a platformReconcileClientAdapter) GetVMStatus(ctx context.Context, node string, vmid int) (ProxmoxVMStatus, error) {
+	status, err := a.client.GetVMStatus(ctx, node, vmid)
+	if err != nil {
+		return ProxmoxVMStatus{}, err
+	}
+
+	return ProxmoxVMStatus{Status: status.Status}, nil
+}

worker/internal/tasks/proxmox_reconcile_test.go

@@ -0,0 +1,124 @@
+package tasks
+
+import (
+	"context"
+	"log/slog"
+	"testing"
+)
+
+func TestProxmoxReconcileHandlerWritesAuditForDrift(t *testing.T) {
+	client := &stubReconcileClient{status: ProxmoxVMStatus{Status: "stopped"}}
+	auditWriter := &stubAuditWriter{}
+	handler := NewProxmoxReconcileHandler(
+		&stubReconcileClusterStore{clusterIDs: []string{"cluster-1"}},
+		&stubReconcileVMStore{vms: []ReconcileVM{{
+			ID:          "vm-1",
+			TenantID:    "tenant-1",
+			Node:        "pve",
+			ProxmoxVMID: 100,
+			Status:      "running",
+		}}},
+		&stubReconcileClientResolver{client: client},
+		auditWriter,
+		slog.Default(),
+	)
+
+	if err := handler.ProcessTask(context.Background(), NewProxmoxReconcileAllTask()); err != nil {
+		t.Fatalf("ProcessTask() error = %v", err)
+	}
+
+	if client.node != "pve" {
+		t.Fatalf("client node = %q, want pve", client.node)
+	}
+	if auditWriter.event.Action != "vm.reconcile.drift" {
+		t.Fatalf("audit action = %q, want vm.reconcile.drift", auditWriter.event.Action)
+	}
+	if auditWriter.event.Metadata["db_status"] != "running" {
+		t.Fatalf("audit db_status = %v, want running", auditWriter.event.Metadata["db_status"])
+	}
+	if auditWriter.event.Metadata["proxmox_status"] != "stopped" {
+		t.Fatalf("audit proxmox_status = %v, want stopped", auditWriter.event.Metadata["proxmox_status"])
+	}
+}
+
+func TestProxmoxReconcileHandlerSkipsMatchingStatus(t *testing.T) {
+	auditWriter := &stubAuditWriter{}
+	handler := NewProxmoxReconcileHandler(
+		&stubReconcileClusterStore{clusterIDs: []string{"cluster-1"}},
+		&stubReconcileVMStore{vms: []ReconcileVM{{
+			ID:          "vm-1",
+			TenantID:    "tenant-1",
+			Node:        "pve",
+			ProxmoxVMID: 100,
+			Status:      "running",
+		}}},
+		&stubReconcileClientResolver{client: &stubReconcileClient{status: ProxmoxVMStatus{Status: "running"}}},
+		auditWriter,
+		slog.Default(),
+	)
+
+	if err := handler.ProcessTask(context.Background(), NewProxmoxReconcileAllTask()); err != nil {
+		t.Fatalf("ProcessTask() error = %v", err)
+	}
+
+	if auditWriter.event.Action != "" {
+		t.Fatalf("audit action = %q, want empty", auditWriter.event.Action)
+	}
+}
+
+func TestNormalizeProxmoxVMStatus(t *testing.T) {
+	tests := []struct {
+		name   string
+		status string
+		want   string
+	}{
+		{name: "running", status: "running", want: "running"},
+		{name: "stopped", status: "stopped", want: "stopped"},
+		{name: "paused", status: "paused", want: "suspended"},
+		{name: "unknown", status: "unknown", want: ""},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := normalizeProxmoxVMStatus(test.status); got != test.want {
+				t.Fatalf("normalizeProxmoxVMStatus() = %q, want %q", got, test.want)
+			}
+		})
+	}
+}
+
+type stubReconcileClusterStore struct {
+	clusterIDs []string
+}
+
+func (s *stubReconcileClusterStore) ListActiveClusterIDs(context.Context) ([]string, error) {
+	return s.clusterIDs, nil
+}
+
+type stubReconcileVMStore struct {
+	vms []ReconcileVM
+}
+
+func (s *stubReconcileVMStore) ListClusterVMs(context.Context, string) ([]ReconcileVM, error) {
+	return s.vms, nil
+}
+
+type stubReconcileClientResolver struct {
+	client ProxmoxReconcileClient
+}
+
+func (s *stubReconcileClientResolver) ResolveReconcileClient(context.Context, string) (ProxmoxReconcileClient, error) {
+	return s.client, nil
+}
+
+type stubReconcileClient struct {
+	node   string
+	vmid   int
+	status ProxmoxVMStatus
+}
+
+func (s *stubReconcileClient) GetVMStatus(_ context.Context, node string, vmid int) (ProxmoxVMStatus, error) {
+	s.node = node
+	s.vmid = vmid
+	return s.status, nil
+}