edgeguard-native/internal/cluster/heartbeat.go

package cluster

// Phase-3.2: periodischer Heartbeat + Stale-Sweeper.
//
// Hintergrund: EnsureSelfRegistered schreibt last_seen einmal beim
// API-Boot. Ohne periodisches Re-Schreiben friert last_seen auf den
// Boot-Zeitpunkt ein — Peers (im Multi-Node-Setup) hätten keinen Weg
// zu erkennen ob dieser Node noch lebt. Die Heartbeat-Goroutine in der
// API bumpt das alle 30s; der Scheduler räumt mit SweepStaleNodes Peers
// die länger als <threshold> nicht gemeldet haben auf status='offline'.
//
// Single-Node-Effekt: Cluster-UI zeigt korrekt "last seen 12s" statt
// "last seen 3h" weil last_seen frisch ist. UI-Drift-Banner-Logik im
// ClusterHandler.Status nutzt die selben Felder.

import (
	"context"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

// Heartbeat bumpt last_seen + status='online' für die eigene Node-Row
// und aktualisiert version + config_hash. Idempotent. UPDATE-only — die
// Row muss existieren (wird via EnsureSelfRegistered beim Boot angelegt).
//
// Hash-Berechnung läuft synchron — typisch <50ms auf einer realistischen
// DB-Größe; falls die compute-SQL fehlt (Migration im Flux) wird der
// vorhandene config_hash via COALESCE behalten.
func Heartbeat(ctx context.Context, pool *pgxpool.Pool, localID, version string) error {
	if pool == nil || localID == "" {
		return nil
	}
	hash, _ := ComputeConfigHash(ctx, pool)
	_, err := pool.Exec(ctx, `
UPDATE ha_nodes SET
    last_seen   = NOW(),
    status      = 'online',
    version     = COALESCE(NULLIF($1, ''), version),
    config_hash = COALESCE(NULLIF($2, ''), config_hash),
    updated_at  = NOW()
WHERE id = $3`, version, hash, localID)
	return err
}

// SweepStaleNodes flippt status='online' → 'offline' für Peers deren
// last_seen älter als threshold ist. Liefert die Anzahl gefliptpter
// Rows zurück (für Logging). Idempotent — markiert keine Rows die
// schon offline sind.
//
// Threshold-Konvention: 4× Heartbeat-Intervall = 2 min bei 30s-Tick.
// Lässt Platz für eine verpasste API-Tick (Restart, GC-Pause, kurzer
// Network-Glitch) ohne false-positive Offline.
func SweepStaleNodes(ctx context.Context, pool *pgxpool.Pool, threshold time.Duration) (int64, error) {
	if pool == nil || threshold <= 0 {
		return 0, nil
	}
	// Wir bauen das Interval als String — pgx kann time.Duration nicht
	// direkt als INTERVAL serialisieren.
	interval := fmt.Sprintf("%d seconds", int(threshold.Seconds()))
	tag, err := pool.Exec(ctx, `
UPDATE ha_nodes SET
    status     = 'offline',
    updated_at = NOW()
WHERE last_seen < NOW() - $1::interval
  AND status = 'online'`, interval)
	if err != nil {
		return 0, err
	}
	return tag.RowsAffected(), nil
}