From 3a3e0e080c5d20875146269cdbb5a215c1e75c13 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 22 Oct 2025 16:10:15 +0000 Subject: [PATCH] Add replication monitoring plumbing and UI Refs #395 --- frontend-modern/src/App.tsx | 8 +- .../components/Proxmox/ProxmoxSectionNav.tsx | 7 +- .../components/Replication/Replication.tsx | 212 ++++++++ frontend-modern/src/stores/websocket.ts | 5 + internal/mock/generator.go | 103 ++++ internal/mock/integration.go | 1 + internal/models/state_snapshot.go | 48 +- pkg/proxmox/client_test.go | 80 +++ pkg/proxmox/replication.go | 457 ++++++++++++++++++ 9 files changed, 898 insertions(+), 23 deletions(-) create mode 100644 frontend-modern/src/components/Replication/Replication.tsx create mode 100644 pkg/proxmox/replication.go diff --git a/frontend-modern/src/App.tsx b/frontend-modern/src/App.tsx index b1056ce..975db56 100644 --- a/frontend-modern/src/App.tsx +++ b/frontend-modern/src/App.tsx @@ -17,6 +17,7 @@ import { getGlobalWebSocketStore } from './stores/websocket-global'; import { Dashboard } from './components/Dashboard/Dashboard'; import StorageComponent from './components/Storage/Storage'; import Backups from './components/Backups/Backups'; +import Replication from './components/Replication/Replication'; import Settings from './components/Settings/Settings'; import { Alerts } from './pages/Alerts'; import { DockerHosts } from './components/Docker/DockerHosts'; @@ -96,12 +97,14 @@ function App() { vms: [], containers: [], dockerHosts: [], + hosts: [], storage: [], cephClusters: [], physicalDisks: [], pbs: [], - pmg: [], - metrics: [], + pmg: [], + replicationJobs: [], + metrics: [], pveBackups: { backupTasks: [], storageBackups: [], @@ -607,6 +610,7 @@ function App() { component={() => } /> + } /> diff --git a/frontend-modern/src/components/Proxmox/ProxmoxSectionNav.tsx b/frontend-modern/src/components/Proxmox/ProxmoxSectionNav.tsx index e91016f..1e7dc43 100644 --- a/frontend-modern/src/components/Proxmox/ProxmoxSectionNav.tsx +++ b/frontend-modern/src/components/Proxmox/ProxmoxSectionNav.tsx @@ -3,7 +3,7 @@ import { createMemo, For } from 'solid-js'; import { useNavigate } from '@solidjs/router'; import { useWebSocket } from '@/App'; -type ProxmoxSection = 'overview' | 'storage' | 'backups' | 'mail'; +type ProxmoxSection = 'overview' | 'storage' | 'replication' | 'backups' | 'mail'; interface ProxmoxSectionNavProps { current: ProxmoxSection; @@ -25,6 +25,11 @@ const allSections: Array<{ label: 'Storage', path: '/proxmox/storage', }, + { + id: 'replication', + label: 'Replication', + path: '/proxmox/replication', + }, { id: 'mail', label: 'Mail Gateway', diff --git a/frontend-modern/src/components/Replication/Replication.tsx b/frontend-modern/src/components/Replication/Replication.tsx new file mode 100644 index 0000000..b2b0e80 --- /dev/null +++ b/frontend-modern/src/components/Replication/Replication.tsx @@ -0,0 +1,212 @@ +import type { Component } from 'solid-js'; +import { Show, For, createMemo } from 'solid-js'; +import { ProxmoxSectionNav } from '@/components/Proxmox/ProxmoxSectionNav'; +import { useWebSocket } from '@/App'; +import { Card } from '@/components/shared/Card'; +import { EmptyState } from '@/components/shared/EmptyState'; +import { formatAbsoluteTime, formatRelativeTime } from '@/utils/format'; +import type { ReplicationJob } from '@/types/api'; + +function formatDuration(durationSeconds?: number, durationHuman?: string): string { + if (durationHuman && durationHuman.trim()) return durationHuman; + if (!durationSeconds || durationSeconds <= 0) return ''; + + const hours = Math.floor(durationSeconds / 3600) + .toString() + .padStart(2, '0'); + const minutes = Math.floor((durationSeconds % 3600) / 60) + .toString() + .padStart(2, '0'); + const seconds = Math.floor(durationSeconds % 60) + .toString() + .padStart(2, '0'); + + return `${hours}:${minutes}:${seconds}`; +} + +function getStatusBadge(job: ReplicationJob) { + const status = (job.status || job.state || '').toLowerCase(); + const lastStatus = (job.lastSyncStatus || '').toLowerCase(); + + if (status.includes('error') || lastStatus.includes('error')) { + return { + tone: 'danger' as const, + label: status || lastStatus || 'Error', + }; + } + + if (status.includes('sync')) { + return { + tone: 'warning' as const, + label: job.status || job.state || 'Syncing', + }; + } + + return { + tone: 'success' as const, + label: job.status || job.state || 'Idle', + }; +} + +function formatRate(limit?: number): string { + if (!limit || limit <= 0) return '—'; + return `${limit.toFixed(0)} MB/s`; +} + +const Replication: Component = () => { + const { state, connected } = useWebSocket(); + + const replicationJobs = createMemo(() => { + const jobs = state.replicationJobs ?? []; + return [...jobs].sort((a, b) => { + if (a.instance !== b.instance) return a.instance.localeCompare(b.instance); + if ((a.guestName || '') !== (b.guestName || '')) { + return (a.guestName || '').localeCompare(b.guestName || ''); + } + return (a.jobId || '').localeCompare(b.jobId || ''); + }); + }); + + return ( +
+ + + + + + + } + title="Connection lost" + description="Unable to connect to the backend server. Attempting to reconnect..." + tone="danger" + /> + + }> + 0} + fallback={ + + + + + } + title="No replication jobs detected" + description="Replication jobs will appear here once configured in Proxmox." + /> + + } + > + +
+ + + + + + + + + + + + + + + + {(job) => { + const badge = getStatusBadge(job); + return ( + + + + + + + + + + + ); + }} + + +
GuestJobSource → TargetLast SyncNext SyncStatusFailuresRate
+
+ {job.guestName || `VM ${job.guestId ?? ''}`} +
+
+ {job.instance} · ID {job.guestId ?? '—'} · {job.guestNode || job.sourceNode || 'Unknown node'} +
+ 		+
{job.jobId || '—'}
+
+ {job.type ? `${job.type.toUpperCase()} · ` : ''}{job.schedule || '*/15'} +
+ 		+
+ {job.sourceNode || '—'} + → + {job.targetNode || '—'} +
+
+ {job.sourceStorage || 'local'} → {job.targetStorage || 'remote'} +
+ 		+ +
{formatAbsoluteTime(job.lastSyncTime!)}
+
+ {formatRelativeTime(job.lastSyncTime!)} + + · + {formatDuration(job.lastSyncDurationSeconds, job.lastSyncDurationHuman)} + +
+ + + Never + + 		+ +
{formatAbsoluteTime(job.nextSyncTime!)}
+
+ {formatRelativeTime(job.nextSyncTime!)} +
+ + + — + + 		+ + {badge.label} + + +
+ {job.error} +
+ + 		+ {job.failCount ?? 0} + + {formatRate(job.rateLimitMbps)} +
+
+ + + +
+ ); +}; + +export default Replication; diff --git a/frontend-modern/src/stores/websocket.ts b/frontend-modern/src/stores/websocket.ts index 8334540..c9b00e4 100644 --- a/frontend-modern/src/stores/websocket.ts +++ b/frontend-modern/src/stores/websocket.ts @@ -26,6 +26,8 @@ export function createWebSocketStore(url: string) { vms: [], containers: [], dockerHosts: [], + hosts: [], + replicationJobs: [], storage: [], cephClusters: [], physicalDisks: [], @@ -350,10 +352,13 @@ export function createWebSocketStore(url: string) { console.log('[WebSocket] Received null dockerHosts, ignoring'); } if (message.data.storage !== undefined) setState('storage', message.data.storage); + if (message.data.hosts !== undefined) setState('hosts', message.data.hosts); if (message.data.cephClusters !== undefined) setState('cephClusters', message.data.cephClusters); if (message.data.pbs !== undefined) setState('pbs', message.data.pbs); if (message.data.pmg !== undefined) setState('pmg', message.data.pmg); + if (message.data.replicationJobs !== undefined) + setState('replicationJobs', message.data.replicationJobs); if (message.data.backups !== undefined) { setState('backups', message.data.backups); if (message.data.backups.pve !== undefined) diff --git a/internal/mock/generator.go b/internal/mock/generator.go index f359b36..f52b395 100644 --- a/internal/mock/generator.go +++ b/internal/mock/generator.go @@ -173,6 +173,7 @@ func GenerateMockData(config MockConfig) models.StateSnapshot { VMs: []models.VM{}, Containers: []models.Container{}, PhysicalDisks: []models.PhysicalDisk{}, + ReplicationJobs: []models.ReplicationJob{}, LastUpdate: time.Now(), ConnectionHealth: make(map[string]bool), Stats: models.Stats{}, @@ -336,6 +337,8 @@ func GenerateMockData(config MockConfig) models.StateSnapshot { PMG: append([]models.PMGBackup(nil), data.PMGBackups...), } + data.ReplicationJobs = generateReplicationJobs(data.Nodes, data.VMs) + // Calculate stats data.Stats.StartTime = time.Now() data.Stats.Uptime = 0 @@ -344,6 +347,106 @@ func GenerateMockData(config MockConfig) models.StateSnapshot { return data } +func generateReplicationJobs(nodes []models.Node, vms []models.VM) []models.ReplicationJob { + if len(nodes) == 0 || len(vms) == 0 { + return []models.ReplicationJob{} + } + + maxJobs := len(vms) + if maxJobs > 8 { + maxJobs = 8 + } + + jobs := make([]models.ReplicationJob, 0, maxJobs) + now := time.Now() + nodeCount := len(nodes) + + for i := 0; i < maxJobs; i++ { + vm := vms[i%len(vms)] + instance := vm.Instance + if instance == "" { + instance = vm.Node + } + + jobNumber := i % 3 + jobID := fmt.Sprintf("%d-%d", vm.VMID, jobNumber) + lastSync := now.Add(-time.Duration(300+rand.Intn(3600)) * time.Second) + nextSync := lastSync.Add(15 * time.Minute) + durationSeconds := 90 + rand.Intn(240) + durationHuman := formatSecondsAsClock(durationSeconds) + status := "idle" + lastStatus := "ok" + errorMessage := "" + failCount := 0 + + roll := rand.Float64() + if roll < 0.1 { + status = "error" + lastStatus = "error" + errorMessage = "last sync timed out" + failCount = 1 + rand.Intn(2) + } else if roll < 0.35 { + status = "syncing" + } + + targetNode := nodes[(i+1)%nodeCount].Name + rate := 80.0 + rand.Float64()*140.0 + + job := models.ReplicationJob{ + ID: fmt.Sprintf("%s-%s", instance, jobID), + Instance: instance, + JobID: jobID, + JobNumber: jobNumber, + Guest: fmt.Sprintf("%d", vm.VMID), + GuestID: vm.VMID, + GuestName: vm.Name, + GuestType: vm.Type, + GuestNode: vm.Node, + SourceNode: vm.Node, + SourceStorage: "local-zfs", + TargetNode: targetNode, + TargetStorage: "replica-zfs", + Schedule: "*/15", + Type: "local", + Enabled: true, + State: status, + Status: status, + LastSyncStatus: lastStatus, + LastSyncTime: ptrTime(lastSync), + LastSyncUnix: lastSync.Unix(), + LastSyncDurationSeconds: durationSeconds, + LastSyncDurationHuman: durationHuman, + NextSyncTime: ptrTime(nextSync), + NextSyncUnix: nextSync.Unix(), + DurationSeconds: durationSeconds, + DurationHuman: durationHuman, + FailCount: failCount, + Error: errorMessage, + RateLimitMbps: ptrFloat64(rate), + LastPolled: now, + } + + jobs = append(jobs, job) + } + + return jobs +} + +func formatSecondsAsClock(totalSeconds int) string { + hours := totalSeconds / 3600 + minutes := (totalSeconds % 3600) / 60 + seconds := totalSeconds % 60 + return fmt.Sprintf("%02d:%02d:%02d", hours, minutes, seconds) +} + +func ptrTime(t time.Time) *time.Time { + return &t +} + +func ptrFloat64(v float64) *float64 { + return &v +} + func generateNodes(config MockConfig) []models.Node { nodes := make([]models.Node, 0, config.NodeCount) diff --git a/internal/mock/integration.go b/internal/mock/integration.go index 6c07a83..7211073 100644 --- a/internal/mock/integration.go +++ b/internal/mock/integration.go @@ -311,6 +311,7 @@ func cloneState(state models.StateSnapshot) models.StateSnapshot { PBSInstances: append([]models.PBSInstance(nil), state.PBSInstances...), PBSBackups: append([]models.PBSBackup(nil), state.PBSBackups...), PMGBackups: append([]models.PMGBackup(nil), state.PMGBackups...), + ReplicationJobs: append([]models.ReplicationJob(nil), state.ReplicationJobs...), Metrics: append([]models.Metric(nil), state.Metrics...), Performance: state.Performance, Stats: state.Stats, diff --git a/internal/models/state_snapshot.go b/internal/models/state_snapshot.go index af82895..237a5ad 100644 --- a/internal/models/state_snapshot.go +++ b/internal/models/state_snapshot.go @@ -4,26 +4,27 @@ import "time" // StateSnapshot represents a snapshot of the state without mutex type StateSnapshot struct { - Nodes []Node `json:"nodes"` - VMs []VM `json:"vms"` - Containers []Container `json:"containers"` - DockerHosts []DockerHost `json:"dockerHosts"` - Storage []Storage `json:"storage"` - CephClusters []CephCluster `json:"cephClusters"` - PhysicalDisks []PhysicalDisk `json:"physicalDisks"` - PBSInstances []PBSInstance `json:"pbs"` - PMGInstances []PMGInstance `json:"pmg"` - PBSBackups []PBSBackup `json:"pbsBackups"` - PMGBackups []PMGBackup `json:"pmgBackups"` - Backups Backups `json:"backups"` - Metrics []Metric `json:"metrics"` - PVEBackups PVEBackups `json:"pveBackups"` - Performance Performance `json:"performance"` - ConnectionHealth map[string]bool `json:"connectionHealth"` - Stats Stats `json:"stats"` - ActiveAlerts []Alert `json:"activeAlerts"` - RecentlyResolved []ResolvedAlert `json:"recentlyResolved"` - LastUpdate time.Time `json:"lastUpdate"` + Nodes []Node `json:"nodes"` + VMs []VM `json:"vms"` + Containers []Container `json:"containers"` + DockerHosts []DockerHost `json:"dockerHosts"` + Storage []Storage `json:"storage"` + CephClusters []CephCluster `json:"cephClusters"` + PhysicalDisks []PhysicalDisk `json:"physicalDisks"` + PBSInstances []PBSInstance `json:"pbs"` + PMGInstances []PMGInstance `json:"pmg"` + PBSBackups []PBSBackup `json:"pbsBackups"` + PMGBackups []PMGBackup `json:"pmgBackups"` + Backups Backups `json:"backups"` + ReplicationJobs []ReplicationJob `json:"replicationJobs"` + Metrics []Metric `json:"metrics"` + PVEBackups PVEBackups `json:"pveBackups"` + Performance Performance `json:"performance"` + ConnectionHealth map[string]bool `json:"connectionHealth"` + Stats Stats `json:"stats"` + ActiveAlerts []Alert `json:"activeAlerts"` + RecentlyResolved []ResolvedAlert `json:"recentlyResolved"` + LastUpdate time.Time `json:"lastUpdate"` } // GetSnapshot returns a snapshot of the current state without mutex @@ -57,6 +58,7 @@ func (s *State) GetSnapshot() StateSnapshot { PBS: pbsBackups, PMG: pmgBackups, }, + ReplicationJobs: append([]ReplicationJob{}, s.ReplicationJobs...), Metrics: append([]Metric{}, s.Metrics...), PVEBackups: pveBackups, Performance: s.Performance, @@ -112,6 +114,11 @@ func (s StateSnapshot) ToFrontend() StateFrontend { cephClusters[i] = cluster.ToFrontend() } + replicationJobs := make([]ReplicationJobFrontend, len(s.ReplicationJobs)) + for i, job := range s.ReplicationJobs { + replicationJobs[i] = job.ToFrontend() + } + return StateFrontend{ Nodes: nodes, VMs: vms, @@ -125,6 +132,7 @@ func (s StateSnapshot) ToFrontend() StateFrontend { PBSBackups: s.PBSBackups, PMGBackups: s.PMGBackups, Backups: s.Backups, + ReplicationJobs: replicationJobs, ActiveAlerts: s.ActiveAlerts, Metrics: make(map[string]any), PVEBackups: s.PVEBackups, diff --git a/pkg/proxmox/client_test.go b/pkg/proxmox/client_test.go index c9e12a1..a4f64e3 100644 --- a/pkg/proxmox/client_test.go +++ b/pkg/proxmox/client_test.go @@ -168,6 +168,86 @@ func TestMemoryStatusEffectiveAvailable(t *testing.T) { } } +func TestMemoryStatusUnmarshalFlexibleValues(t *testing.T) { + tests := []struct { + name string + payload string + want MemoryStatus + }{ + { + name: "numeric strings", + payload: `{"total":"16549875712","used":"6050492416","free":"2467389440","available":"10499383296","buffers":"0","cached":"0","shared":"0"}`, + want: MemoryStatus{ + Total: 16549875712, + Used: 6050492416, + Free: 2467389440, + Available: 10499383296, + Buffers: 0, + Cached: 0, + Shared: 0, + }, + }, + { + name: "scientific notation", + payload: `{"total":1.6549875712e+10,"used":6.050492416e+09,"free":2.46738944e+09,"available":1.0499383296e+10,"buffers":0,"cached":0,"shared":0}`, + want: MemoryStatus{ + Total: 16549875712, + Used: 6050492416, + Free: 2467389440, + Available: 10499383296, + }, + }, + { + name: "float-like strings with spaces", + payload: `{"total":" 8589934592.0 ","used":"3221225472.0","free":"536870912.0","avail":"4831838208.0","buffers":"67108864","cached":"4026531840","shared":"0"}`, + want: MemoryStatus{ + Total: 8589934592, + Used: 3221225472, + Free: 536870912, + Avail: 4831838208, + Buffers: 67108864, + Cached: 4026531840, + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var status MemoryStatus + if err := json.Unmarshal([]byte(tc.payload), &status); err != nil { + t.Fatalf("unexpected error unmarshalling %s payload: %v", tc.name, err) + } + + if status.Total != tc.want.Total { + t.Fatalf("total: got %d, want %d", status.Total, tc.want.Total) + } + if status.Used != tc.want.Used { + t.Fatalf("used: got %d, want %d", status.Used, tc.want.Used) + } + if status.Free != tc.want.Free { + t.Fatalf("free: got %d, want %d", status.Free, tc.want.Free) + } + if status.Available != tc.want.Available { + t.Fatalf("available: got %d, want %d", status.Available, tc.want.Available) + } + if status.Avail != tc.want.Avail { + t.Fatalf("avail: got %d, want %d", status.Avail, tc.want.Avail) + } + if status.Buffers != tc.want.Buffers { + t.Fatalf("buffers: got %d, want %d", status.Buffers, tc.want.Buffers) + } + if status.Cached != tc.want.Cached { + t.Fatalf("cached: got %d, want %d", status.Cached, tc.want.Cached) + } + if status.Shared != tc.want.Shared { + t.Fatalf("shared: got %d, want %d", status.Shared, tc.want.Shared) + } + }) + } +} + + + // TestMemoryStatusEffectiveAvailable_RegressionIssue435 tests the specific scenarios // reported in GitHub issue #435 where memory calculations incorrectly included cache/buffers func TestMemoryStatusEffectiveAvailable_RegressionIssue435(t *testing.T) { diff --git a/pkg/proxmox/replication.go b/pkg/proxmox/replication.go new file mode 100644 index 0000000..ea60990 --- /dev/null +++ b/pkg/proxmox/replication.go @@ -0,0 +1,457 @@ +package proxmox + +import ( + "context" + "encoding/json" + "fmt" + "math" + "strconv" + "strings" + "time" +) + +// ReplicationJob represents the parsed status of a Proxmox storage replication job. +type ReplicationJob struct { + ID string `json:"id"` + Guest string `json:"guest,omitempty"` + GuestID int `json:"guestId,omitempty"` + JobNumber int `json:"jobNumber,omitempty"` + Source string `json:"source,omitempty"` + SourceStorage string `json:"sourceStorage,omitempty"` + Target string `json:"target,omitempty"` + TargetStorage string `json:"targetStorage,omitempty"` + Schedule string `json:"schedule,omitempty"` + Type string `json:"type,omitempty"` + Enabled bool `json:"enabled"` + State string `json:"state,omitempty"` + Status string `json:"status,omitempty"` + LastSyncStatus string `json:"lastSyncStatus,omitempty"` + LastSyncTime *time.Time `json:"lastSyncTime,omitempty"` + LastSyncUnix int64 `json:"lastSyncUnix,omitempty"` + LastSyncDurationSeconds int `json:"lastSyncDurationSeconds,omitempty"` + LastSyncDurationHuman string `json:"lastSyncDurationHuman,omitempty"` + NextSyncTime *time.Time `json:"nextSyncTime,omitempty"` + NextSyncUnix int64 `json:"nextSyncUnix,omitempty"` + DurationSeconds int `json:"durationSeconds,omitempty"` + DurationHuman string `json:"durationHuman,omitempty"` + FailCount int `json:"failCount,omitempty"` + Error string `json:"error,omitempty"` + Comment string `json:"comment,omitempty"` + RemoveJob string `json:"removeJob,omitempty"` + RateLimitMbps *float64 `json:"rateLimitMbps,omitempty"` +} + +// GetReplicationStatus returns the replication jobs configured on a PVE instance. +func (c *Client) GetReplicationStatus(ctx context.Context) ([]ReplicationJob, error) { + resp, err := c.get(ctx, "/cluster/replication") + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var raw struct { + Data []map[string]json.RawMessage `json:"data"` + } + + if err := json.NewDecoder(resp.Body).Decode(&raw); err != nil { + return nil, err + } + + jobs := make([]ReplicationJob, 0, len(raw.Data)) + for _, entry := range raw.Data { + jobs = append(jobs, parseReplicationJob(entry)) + } + + return jobs, nil +} + +func parseReplicationJob(entry map[string]json.RawMessage) ReplicationJob { + job := ReplicationJob{ + Enabled: true, + } + + job.ID = stringFromAny(decodeRaw(entry["id"])) + if job.ID == "" { + job.ID = stringFromAny(decodeRaw(entry["jobid"])) + } + + job.Guest = stringFromAny(decodeRaw(entry["guest"])) + if guestID, ok := intFromAny(decodeRaw(entry["guest"])); ok { + job.GuestID = guestID + } + + if jobNum, ok := intFromAny(decodeRaw(entry["jobnum"])); ok { + job.JobNumber = jobNum + } else if parts := strings.Split(job.ID, "-"); len(parts) == 2 { + if num, err := strconv.Atoi(parts[1]); err == nil { + job.JobNumber = num + } + } + + job.Source = stringFromAny(decodeRaw(entry["source"])) + job.SourceStorage = stringFromAny(firstNonNilRaw(entry, "source-storage", "source_storage")) + job.Target = stringFromAny(decodeRaw(entry["target"])) + job.TargetStorage = stringFromAny(firstNonNilRaw(entry, "target-storage", "target_storage")) + job.Schedule = stringFromAny(decodeRaw(entry["schedule"])) + job.Type = stringFromAny(decodeRaw(entry["type"])) + job.Comment = stringFromAny(decodeRaw(entry["comment"])) + job.RemoveJob = stringFromAny(firstNonNilRaw(entry, "remove_job", "remove-job")) + + if enabled, ok := boolFromAny(decodeRaw(entry["enabled"])); ok { + job.Enabled = enabled + } + if disabled, ok := boolFromAny(decodeRaw(entry["disable"])); ok && disabled { + job.Enabled = false + } + if active, ok := boolFromAny(decodeRaw(entry["active"])); ok && !active { + job.Enabled = false + } + + job.State = stringFromAny(decodeRaw(entry["state"])) + job.Status = stringFromAny(decodeRaw(entry["status"])) + if job.Status == "" && job.State != "" { + job.Status = job.State + } + + job.LastSyncStatus = stringFromAny(firstNonNilRaw(entry, "last_sync_status", "last-sync-status", "last_sync_state", "last-sync-state")) + job.Error = stringFromAny(decodeRaw(entry["error"])) + if job.Error == "" { + job.Error = stringFromAny(decodeRaw(entry["last_sync_error"])) + } + job.FailCount, _ = intFromAny(firstNonNilRaw(entry, "fail_count", "fail-count")) + + if seconds, human := parseDurationSeconds(decodeRaw(entry["last_sync_duration"])); seconds > 0 { + job.LastSyncDurationSeconds = seconds + job.LastSyncDurationHuman = human + } else if seconds, human := parseDurationSeconds(firstNonNilRaw(entry, "last-sync-duration", "last_sync_duration_sec")); seconds > 0 { + job.LastSyncDurationSeconds = seconds + job.LastSyncDurationHuman = human + } + + if seconds, human := parseDurationSeconds(decodeRaw(entry["duration"])); seconds > 0 { + job.DurationSeconds = seconds + job.DurationHuman = human + } + + if t, unix := parseReplicationTime(decodeRaw(entry["last_sync"])); t != nil { + job.LastSyncTime = t + job.LastSyncUnix = unix + } else if t, unix := parseReplicationTime(firstNonNilRaw(entry, "last-sync", "last_sync_time")); t != nil { + job.LastSyncTime = t + job.LastSyncUnix = unix + } + + if t, unix := parseReplicationTime(decodeRaw(entry["next_sync"])); t != nil { + job.NextSyncTime = t + job.NextSyncUnix = unix + } else if t, unix := parseReplicationTime(firstNonNilRaw(entry, "next-sync", "next_sync_time")); t != nil { + job.NextSyncTime = t + job.NextSyncUnix = unix + } + + if rate, ok := floatFromAny(decodeRaw(entry["rate"])); ok { + job.RateLimitMbps = copyFloat(rate) + } + + return job +} + +func decodeRaw(raw json.RawMessage) interface{} { + if raw == nil { + return nil + } + var value interface{} + if err := json.Unmarshal(raw, &value); err != nil { + return nil + } + return value +} + +func firstNonNilRaw(entry map[string]json.RawMessage, keys ...string) interface{} { + for _, key := range keys { + if value, ok := entry[key]; ok && value != nil { + return decodeRaw(value) + } + } + return nil +} + +func stringFromAny(value interface{}) string { + switch v := value.(type) { + case nil: + return "" + case string: + return strings.TrimSpace(v) + case json.Number: + return strings.TrimSpace(v.String()) + case float64: + if math.IsNaN(v) || math.IsInf(v, 0) { + return "" + } + return strings.TrimSpace(strconv.FormatFloat(v, 'f', -1, 64)) + case float32: + return strings.TrimSpace(strconv.FormatFloat(float64(v), 'f', -1, 32)) + case int: + return strconv.Itoa(v) + case int64: + return strconv.FormatInt(v, 10) + case int32: + return strconv.FormatInt(int64(v), 10) + case uint: + return strconv.FormatUint(uint64(v), 10) + case uint64: + return strconv.FormatUint(v, 10) + case uint32: + return strconv.FormatUint(uint64(v), 10) + case bool: + if v { + return "true" + } + return "false" + default: + return strings.TrimSpace(fmt.Sprint(v)) + } +} + +func intFromAny(value interface{}) (int, bool) { + switch v := value.(type) { + case nil: + return 0, false + case int: + return v, true + case int8: + return int(v), true + case int16: + return int(v), true + case int32: + return int(v), true + case int64: + return int(v), true + case uint: + return int(v), true + case uint8: + return int(v), true + case uint16: + return int(v), true + case uint32: + return int(v), true + case uint64: + return int(v), true + case float32: + if math.IsNaN(float64(v)) || math.IsInf(float64(v), 0) { + return 0, false + } + return int(math.Round(float64(v))), true + case float64: + if math.IsNaN(v) || math.IsInf(v, 0) { + return 0, false + } + return int(math.Round(v)), true + case json.Number: + if i, err := v.Int64(); err == nil { + return int(i), true + } + if f, err := v.Float64(); err == nil && !math.IsNaN(f) && !math.IsInf(f, 0) { + return int(math.Round(f)), true + } + case string: + s := strings.TrimSpace(v) + if s == "" { + return 0, false + } + if i, err := strconv.ParseInt(s, 10, 64); err == nil { + return int(i), true + } + if f, err := strconv.ParseFloat(s, 64); err == nil && !math.IsNaN(f) && !math.IsInf(f, 0) { + return int(math.Round(f)), true + } + } + return 0, false +} + +func boolFromAny(value interface{}) (bool, bool) { + switch v := value.(type) { + case nil: + return false, false + case bool: + return v, true + case int, int8, int16, int32, int64: + i, ok := intFromAny(v) + return i != 0, ok + case uint, uint8, uint16, uint32, uint64: + i, ok := intFromAny(v) + return i != 0, ok + case float32, float64: + i, ok := intFromAny(v) + return i != 0, ok + case json.Number: + i, ok := intFromAny(v) + return i != 0, ok + case string: + s := strings.TrimSpace(strings.ToLower(v)) + switch s { + case "true", "yes", "1", "on", "enabled": + return true, true + case "false", "no", "0", "off", "disabled": + return false, true + } + } + return false, false +} + +func floatFromAny(value interface{}) (float64, bool) { + switch v := value.(type) { + case nil: + return 0, false + case float64: + if math.IsNaN(v) || math.IsInf(v, 0) { + return 0, false + } + return v, true + case float32: + if math.IsNaN(float64(v)) || math.IsInf(float64(v), 0) { + return 0, false + } + return float64(v), true + case int, int8, int16, int32, int64: + i, ok := intFromAny(v) + return float64(i), ok + case uint, uint8, uint16, uint32, uint64: + i, ok := intFromAny(v) + return float64(i), ok + case json.Number: + if f, err := v.Float64(); err == nil && !math.IsNaN(f) && !math.IsInf(f, 0) { + return f, true + } + case string: + s := strings.TrimSpace(v) + if s == "" { + return 0, false + } + if f, err := strconv.ParseFloat(s, 64); err == nil && !math.IsNaN(f) && !math.IsInf(f, 0) { + return f, true + } + } + return 0, false +} + +func parseReplicationTime(value interface{}) (*time.Time, int64) { + switch v := value.(type) { + case nil: + return nil, 0 + case time.Time: + t := v.UTC() + return &t, t.Unix() + case *time.Time: + if v == nil { + return nil, 0 + } + t := v.UTC() + return &t, t.Unix() + case int, int32, int64, uint, uint32, uint64: + i, ok := intFromAny(v) + if !ok || i <= 0 { + return nil, 0 + } + t := time.Unix(int64(i), 0).UTC() + return &t, t.Unix() + case float32, float64, json.Number: + f, ok := floatFromAny(v) + if !ok || f <= 0 { + return nil, 0 + } + t := time.Unix(int64(math.Round(f)), 0).UTC() + return &t, t.Unix() + case string: + s := strings.TrimSpace(v) + if s == "" || strings.EqualFold(s, "n/a") || strings.EqualFold(s, "pending") || strings.EqualFold(s, "-") { + return nil, 0 + } + + if i, err := strconv.ParseInt(s, 10, 64); err == nil { + if i <= 0 { + return nil, 0 + } + t := time.Unix(i, 0).UTC() + return &t, t.Unix() + } + + layouts := []string{ + time.RFC3339, + "2006-01-02 15:04:05", + "2006-01-02 15:04:05 -0700", + "2006-01-02 15:04:05 MST", + "2006-01-02T15:04:05", + } + + for _, layout := range layouts { + if t, err := time.ParseInLocation(layout, s, time.Local); err == nil { + t = t.UTC() + return &t, t.Unix() + } + } + } + + return nil, 0 +} + +func parseDurationSeconds(value interface{}) (int, string) { + if value == nil { + return 0, "" + } + + switch v := value.(type) { + case int, int32, int64, uint, uint32, uint64: + if secs, ok := intFromAny(v); ok && secs >= 0 { + return secs, strconv.Itoa(secs) + } + case float32, float64, json.Number: + if secs, ok := floatFromAny(v); ok && secs >= 0 { + return int(math.Round(secs)), strconv.FormatFloat(secs, 'f', -1, 64) + } + case string: + s := strings.TrimSpace(v) + if s == "" { + return 0, "" + } + if strings.Contains(s, ":") { + if secs, ok := parseHHMMSSToSeconds(s); ok { + return secs, s + } + } + if secs, err := strconv.ParseFloat(s, 64); err == nil && secs >= 0 { + return int(math.Round(secs)), s + } + } + + return 0, stringFromAny(value) +} + +func parseHHMMSSToSeconds(value string) (int, bool) { + parts := strings.Split(value, ":") + if len(parts) < 2 || len(parts) > 3 { + return 0, false + } + + total := 0 + multipliers := []int{3600, 60, 1} + + for i := 0; i < len(parts); i++ { + idx := len(multipliers) - len(parts) + i + part := strings.TrimSpace(parts[i]) + if part == "" { + return 0, false + } + val, err := strconv.Atoi(part) + if err != nil { + return 0, false + } + total += val * multipliers[idx] + } + + return total, true +} + +func copyFloat(value float64) *float64 { + return &value +}