proxui/worker/internal/tasks/proxmox_reconcile.go

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
}