updates

internal/hub/probes.go

@@ -1,8 +1,6 @@
 package hub
 
 import (
-	"strconv"
-
 	"github.com/henrygd/beszel/internal/entities/probe"
 	"github.com/henrygd/beszel/internal/hub/systems"
 	"github.com/pocketbase/pocketbase/core"
@@ -10,9 +8,7 @@ import (
 
 // generateProbeID creates a stable hash ID for a probe based on its configuration and the system it belongs to.
 func generateProbeID(systemId string, config probe.Config) string {
-	intervalStr := strconv.FormatUint(uint64(config.Interval), 10)
-	portStr := strconv.FormatUint(uint64(config.Port), 10)
-	return systems.MakeStableHashId(systemId, config.Protocol, config.Target, portStr, intervalStr)
+	return systems.MakeStableHashId(systemId, config.Target, config.Protocol)
 }
 
 // bindNetworkProbesEvents keeps probe records and agent probe state in sync.
@@ -27,7 +23,7 @@ func bindNetworkProbesEvents(hub *Hub) {
 	})
 
 	// sync probe to agent on creation and persist the first result immediately when available
-	hub.OnRecordCreateRequest("network_probes").BindFunc(func(e *core.RecordRequestEvent) error {
+	hub.OnRecordAfterCreateSuccess("network_probes").BindFunc(func(e *core.RecordEvent) error {
 		err := e.Next()
 		if err != nil {
 			return err
@@ -47,10 +43,24 @@ func bindNetworkProbesEvents(hub *Hub) {
 		if err := e.App.SaveNoValidate(e.Record); err != nil {
 			hub.Logger().Warn("failed to save initial probe result", "system", e.Record.GetString("system"), "probe", e.Record.Id, "err", err)
 		}
-		return nil
+		return e.Next()
 	})
 
+	// On API update requests, if the probe config changed in a way that requires a new ID, we will create a new
+	// record with the new ID and delete the old one. Otherwise, we will just update the existing probe on the agent.
 	hub.OnRecordUpdateRequest("network_probes").BindFunc(func(e *core.RecordRequestEvent) error {
+		systemID := e.Record.GetString("system")
+		ID := generateProbeID(systemID, *probeConfigFromRecord(e.Record))
+		if ID != e.Record.Id {
+			newRecord := copyProbeToNewRecord(e.Record, ID)
+			if err := e.App.Save(newRecord); err != nil {
+				return err
+			}
+			if err := e.App.Delete(e.Record); err != nil {
+				return err
+			}
+			return nil
+		}
 		err := e.Next()
 		if err != nil {
 			return err
@@ -67,15 +77,11 @@ func bindNetworkProbesEvents(hub *Hub) {
 	})
 
 	// sync probe to agent on delete
-	hub.OnRecordDeleteRequest("network_probes").BindFunc(func(e *core.RecordRequestEvent) error {
-		err := e.Next()
-		if err != nil {
-			return err
-		}
+	hub.OnRecordAfterDeleteSuccess("network_probes").BindFunc(func(e *core.RecordEvent) error {
 		if err := hub.deleteNetworkProbe(e.Record); err != nil {
 			hub.Logger().Warn("failed to delete probe on agent", "system", e.Record.GetString("system"), "probe", e.Record.Id, "err", err)
 		}
-		return nil
+		return e.Next()
 	})
 }
 
@@ -99,6 +105,17 @@ func setProbeResultFields(record *core.Record, result probe.Result) {
 	record.Set("loss1h", result.Get(4))
 }
 
+// copyProbeToNewRecord creates a new record with the same field values as the old one.
+// This is used when the probe config changes in a way that requires a new ID, so we need
+// to create a new record with the new ID and delete the old one.
+func copyProbeToNewRecord(oldRecord *core.Record, newID string) *core.Record {
+	collection := oldRecord.Collection()
+	newRecord := core.NewRecord(collection)
+	newRecord.Load(oldRecord.FieldsData())
+	newRecord.Set("id", newID)
+	return newRecord
+}
+
 // upsertNetworkProbe applies the record's probe config to the target system.
 func (h *Hub) upsertNetworkProbe(record *core.Record, runNow bool) (*probe.Result, error) {
 	systemID := record.GetString("system")

internal/hub/probes_test.go

@@ -23,7 +23,7 @@ func TestGenerateProbeID(t *testing.T) {
 				Port:     80,
 				Interval: 60,
 			},
-			expected: "d5f27931",
+			expected: "a20a5827",
 		},
 		{
 			name:     "HTTP probe on example.com with different system ID",
@@ -34,7 +34,7 @@ func TestGenerateProbeID(t *testing.T) {
 				Port:     80,
 				Interval: 60,
 			},
-			expected: "6f8b17f1",
+			expected: "ab602ae7",
 		},
 		{
 			name:     "Same probe, different interval",
@@ -45,7 +45,7 @@ func TestGenerateProbeID(t *testing.T) {
 				Port:     80,
 				Interval: 120,
 			},
-			expected: "6d4baf8",
+			expected: "ab602ae7",
 		},
 		{
 			name:     "ICMP probe on 1.1.1.1",
@@ -56,7 +56,7 @@ func TestGenerateProbeID(t *testing.T) {
 				Port:     0,
 				Interval: 10,
 			},
-			expected: "80b5836b",
+			expected: "6d13a4a4",
 		}, {
 			name:     "ICMP probe on 1.1.1.1 with different system ID",
 			systemID: "sys4567",
@@ -66,7 +66,7 @@ func TestGenerateProbeID(t *testing.T) {
 				Port:     0,
 				Interval: 10,
 			},
-			expected: "a6652680",
+			expected: "ddd6c81",
 		},
 	}
 

internal/hub/systems/system.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"hash/fnv"
-	"log/slog"
 	"math/rand"
 	"net"
 	"strings"
@@ -325,7 +324,6 @@ func updateNetworkProbesRecords(app core.App, data map[string]probe.Result, syst
 	// If realtime updates are active, we save via PocketBase records to trigger realtime events.
 	// Otherwise we can do a more efficient direct update via SQL
 	realtimeActive := utils.RealtimeActiveForCollection(app, collectionName, func(filterQuery string) bool {
-		slog.Info("Checking realtime subscription filter for network probes", "filterQuery", filterQuery)
 		return !strings.Contains(filterQuery, "system") || strings.Contains(filterQuery, systemId)
 	})
 
@@ -335,35 +333,10 @@ func updateNetworkProbesRecords(app core.App, data map[string]probe.Result, syst
 	if !realtimeActive {
 		db = app.DB()
 		nowString = time.Now().UTC().Format(types.DefaultDateLayout)
-		sql := fmt.Sprintf("UPDATE %s SET resAvg={:res}, resMin1h={:resMin1h}, resMax1h={:resMax1h}, resAvg1h={:resAvg1h}, loss1h={:loss1h}, updated={:updated} WHERE id={:id}", collectionName)
+		sql := fmt.Sprintf("UPDATE %s SET res={:res}, resMin1h={:resMin1h}, resMax1h={:resMax1h}, resAvg1h={:resAvg1h}, loss1h={:loss1h}, updated={:updated} WHERE id={:id}", collectionName)
 		updateQuery = db.NewQuery(sql)
 	}
 
-	// insert network probe stats records
-	switch realtimeActive {
-	case true:
-		collection, _ := app.FindCachedCollectionByNameOrId("network_probe_stats")
-		record := core.NewRecord(collection)
-		record.Set("system", systemId)
-		record.Set("stats", data)
-		record.Set("type", "1m")
-		err = app.SaveNoValidate(record)
-	default:
-		if dataJSON, marshalErr := json.Marshal(data); marshalErr == nil {
-			sql := "INSERT INTO network_probe_stats (system, stats, type, created) VALUES ({:system}, {:stats}, {:type}, {:created})"
-			insertQuery := db.NewQuery(sql)
-			_, err = insertQuery.Bind(dbx.Params{
-				"system":  systemId,
-				"stats":   dataJSON,
-				"type":    "1m",
-				"created": nowString,
-			}).Execute()
-		}
-	}
-	if err != nil {
-		app.Logger().Error("Failed to update probe stats", "system", systemId, "err", err)
-	}
-
 	// update network_probes records
 	for id, values := range data {
 		switch realtimeActive {
@@ -394,6 +367,31 @@ func updateNetworkProbesRecords(app core.App, data map[string]probe.Result, syst
 		}
 	}
 
+	// insert network probe stats records
+	switch realtimeActive {
+	case true:
+		collection, _ := app.FindCachedCollectionByNameOrId("network_probe_stats")
+		record := core.NewRecord(collection)
+		record.Set("system", systemId)
+		record.Set("stats", data)
+		record.Set("type", "1m")
+		err = app.SaveNoValidate(record)
+	default:
+		var statsJson types.JSONRaw
+		if err := statsJson.Scan(data); err == nil {
+			insertQuery := db.NewQuery("INSERT INTO network_probe_stats (system, stats, type, created) VALUES ({:system}, {:stats}, {:type}, {:created})")
+			_, err = insertQuery.Bind(dbx.Params{
+				"system":  systemId,
+				"stats":   statsJson,
+				"type":    "1m",
+				"created": nowString,
+			}).Execute()
+		}
+	}
+	if err != nil {
+		app.Logger().Error("Failed to update probe stats", "system", systemId, "err", err)
+	}
+
 	return nil
 }