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 }