refactor: simplify/improve status alert handling (#1519)

also adds new functionality to restore any pending down alerts
that were lost by hub restart before creation

Author: henrygd

internal/alerts/alerts.go

@@ -21,8 +21,7 @@ type hubLike interface {
 
 type AlertManager struct {
 	hub           hubLike
-	alertQueue    chan alertTask
-	stopChan      chan struct{}
+	stopOnce      sync.Once
 	pendingAlerts sync.Map
 }
 
@@ -98,12 +97,9 @@ var supportsTitle = map[string]struct{}{
 // NewAlertManager creates a new AlertManager instance.
 func NewAlertManager(app hubLike) *AlertManager {
 	am := &AlertManager{
-		hub:        app,
-		alertQueue: make(chan alertTask, 5),
-		stopChan:   make(chan struct{}),
+		hub: app,
 	}
 	am.bindEvents()
-	go am.startWorker()
 	return am
 }
 
@@ -112,6 +108,16 @@ func (am *AlertManager) bindEvents() {
 	am.hub.OnRecordAfterUpdateSuccess("alerts").BindFunc(updateHistoryOnAlertUpdate)
 	am.hub.OnRecordAfterDeleteSuccess("alerts").BindFunc(resolveHistoryOnAlertDelete)
 	am.hub.OnRecordAfterUpdateSuccess("smart_devices").BindFunc(am.handleSmartDeviceAlert)
+
+	am.hub.OnServe().BindFunc(func(e *core.ServeEvent) error {
+		if err := resolveStatusAlerts(e.App); err != nil {
+			e.App.Logger().Error("Failed to resolve stale status alerts", "err", err)
+		}
+		if err := am.restorePendingStatusAlerts(); err != nil {
+			e.App.Logger().Error("Failed to restore pending status alerts", "err", err)
+		}
+		return e.Next()
+	})
 }
 
 // IsNotificationSilenced checks if a notification should be silenced based on configured quiet hours

internal/alerts/alerts_quiet_hours_test.go

@@ -49,7 +49,7 @@ func TestAlertSilencedOneTime(t *testing.T) {
 
 	// Get alert manager
 	am := alerts.NewAlertManager(hub)
-	defer am.StopWorker()
+	defer am.Stop()
 
 	// Test that alert is silenced
 	silenced := am.IsNotificationSilenced(user.Id, system.Id)
@@ -106,7 +106,7 @@ func TestAlertSilencedDaily(t *testing.T) {
 
 	// Get alert manager
 	am := alerts.NewAlertManager(hub)
-	defer am.StopWorker()
+	defer am.Stop()
 
 	// Get current hour and create a window that includes current time
 	now := time.Now().UTC()
@@ -170,7 +170,7 @@ func TestAlertSilencedDailyMidnightCrossing(t *testing.T) {
 
 	// Get alert manager
 	am := alerts.NewAlertManager(hub)
-	defer am.StopWorker()
+	defer am.Stop()
 
 	// Create a window that crosses midnight: 22:00 - 02:00
 	startTime := time.Date(2000, 1, 1, 22, 0, 0, 0, time.UTC)
@@ -211,7 +211,7 @@ func TestAlertSilencedGlobal(t *testing.T) {
 
 	// Get alert manager
 	am := alerts.NewAlertManager(hub)
-	defer am.StopWorker()
+	defer am.Stop()
 
 	// Create a global quiet hours window (no system specified)
 	now := time.Now().UTC()
@@ -250,7 +250,7 @@ func TestAlertSilencedSystemSpecific(t *testing.T) {
 
 	// Get alert manager
 	am := alerts.NewAlertManager(hub)
-	defer am.StopWorker()
+	defer am.Stop()
 
 	// Create a system-specific quiet hours window for system1 only
 	now := time.Now().UTC()
@@ -296,7 +296,7 @@ func TestAlertSilencedMultiUser(t *testing.T) {
 
 	// Get alert manager
 	am := alerts.NewAlertManager(hub)
-	defer am.StopWorker()
+	defer am.Stop()
 
 	// Create a quiet hours window for user1 only
 	now := time.Now().UTC()
@@ -417,7 +417,7 @@ func TestAlertSilencedNoWindows(t *testing.T) {
 
 	// Get alert manager
 	am := alerts.NewAlertManager(hub)
-	defer am.StopWorker()
+	defer am.Stop()
 
 	// Without any quiet hours windows, alert should NOT be silenced
 	silenced := am.IsNotificationSilenced(user.Id, system.Id)

internal/alerts/alerts_status.go

@@ -9,63 +9,25 @@ import (
 	"github.com/pocketbase/pocketbase/core"
 )
 
-type alertTask struct {
-	action      string // "schedule" or "cancel"
-	systemName  string
-	alertRecord *core.Record
-	delay       time.Duration
-}
-
 type alertInfo struct {
 	systemName  string
 	alertRecord *core.Record
 	expireTime  time.Time
+	timer       *time.Timer
 }
 
-// startWorker is a long-running goroutine that processes alert tasks
-// every x seconds. It must be running to process status alerts.
-func (am *AlertManager) startWorker() {
-	processPendingAlerts := time.Tick(15 * time.Second)
-
-	// check for status alerts that are not resolved when system comes up
-	// (can be removed if we figure out core bug in #1052)
-	checkStatusAlerts := time.Tick(561 * time.Second)
-
-	for {
-		select {
-		case <-am.stopChan:
-			return
-		case task := <-am.alertQueue:
-			switch task.action {
-			case "schedule":
-				am.pendingAlerts.Store(task.alertRecord.Id, &alertInfo{
-					systemName:  task.systemName,
-					alertRecord: task.alertRecord,
-					expireTime:  time.Now().Add(task.delay),
-				})
-			case "cancel":
-				am.pendingAlerts.Delete(task.alertRecord.Id)
+// Stop cancels all pending status alert timers.
+func (am *AlertManager) Stop() {
+	am.stopOnce.Do(func() {
+		am.pendingAlerts.Range(func(key, value any) bool {
+			info := value.(*alertInfo)
+			if info.timer != nil {
+				info.timer.Stop()
 			}
-		case <-checkStatusAlerts:
-			resolveStatusAlerts(am.hub)
-		case <-processPendingAlerts:
-			// Check for expired alerts every tick
-			now := time.Now()
-			for key, value := range am.pendingAlerts.Range {
-				info := value.(*alertInfo)
-				if now.After(info.expireTime) {
-					// Downtime delay has passed, process alert
-					am.sendStatusAlert("down", info.systemName, info.alertRecord)
-					am.pendingAlerts.Delete(key)
-				}
-			}
-		}
-	}
-}
-
-// StopWorker shuts down the AlertManager.worker goroutine
-func (am *AlertManager) StopWorker() {
-	close(am.stopChan)
+			am.pendingAlerts.Delete(key)
+			return true
+		})
+	})
 }
 
 // HandleStatusAlerts manages the logic when system status changes.
@@ -103,38 +65,43 @@ func (am *AlertManager) getSystemStatusAlerts(systemID string) ([]*core.Record,
 	return alertRecords, nil
 }
 
-// Schedules delayed "down" alerts for each alert record.
+// handleSystemDown manages the logic when a system status changes to "down". It schedules pending alerts for each alert record.
 func (am *AlertManager) handleSystemDown(systemName string, alertRecords []*core.Record) {
 	for _, alertRecord := range alertRecords {
-		// Continue if alert is already scheduled
-		if _, exists := am.pendingAlerts.Load(alertRecord.Id); exists {
-			continue
-		}
-		// Schedule by adding to queue
 		min := max(1, alertRecord.GetInt("min"))
-		am.alertQueue <- alertTask{
-			action:      "schedule",
-			systemName:  systemName,
-			alertRecord: alertRecord,
-			delay:       time.Duration(min) * time.Minute,
-		}
+		am.schedulePendingStatusAlert(systemName, alertRecord, time.Duration(min)*time.Minute)
 	}
 }
 
+// schedulePendingStatusAlert sets up a timer to send a "down" alert after the specified delay if the system is still down.
+// It returns true if the alert was scheduled, or false if an alert was already pending for the given alert record.
+func (am *AlertManager) schedulePendingStatusAlert(systemName string, alertRecord *core.Record, delay time.Duration) bool {
+	alert := &alertInfo{
+		systemName:  systemName,
+		alertRecord: alertRecord,
+		expireTime:  time.Now().Add(delay),
+	}
+
+	storedAlert, loaded := am.pendingAlerts.LoadOrStore(alertRecord.Id, alert)
+	if loaded {
+		return false
+	}
+
+	stored := storedAlert.(*alertInfo)
+	stored.timer = time.AfterFunc(time.Until(stored.expireTime), func() {
+		am.processPendingAlert(alertRecord.Id)
+	})
+	return true
+}
+
 // handleSystemUp manages the logic when a system status changes to "up".
 // It cancels any pending alerts and sends "up" alerts.
 func (am *AlertManager) handleSystemUp(systemName string, alertRecords []*core.Record) {
 	for _, alertRecord := range alertRecords {
-		alertRecordID := alertRecord.Id
 		// If alert exists for record, delete and continue (down alert not sent)
-		if _, exists := am.pendingAlerts.Load(alertRecordID); exists {
-			am.alertQueue <- alertTask{
-				action:      "cancel",
-				alertRecord: alertRecord,
-			}
+		if am.cancelPendingAlert(alertRecord.Id) {
 			continue
 		}
-		// No alert scheduled for this record, send "up" alert only if "down" was triggered
 		if !alertRecord.GetBool("triggered") {
 			continue
 		}
@@ -144,6 +111,36 @@ func (am *AlertManager) handleSystemUp(systemName string, alertRecords []*core.R
 	}
 }
 
+// cancelPendingAlert stops the timer and removes the pending alert for the given alert ID. Returns true if a pending alert was found and cancelled.
+func (am *AlertManager) cancelPendingAlert(alertID string) bool {
+	value, loaded := am.pendingAlerts.LoadAndDelete(alertID)
+	if !loaded {
+		return false
+	}
+
+	info := value.(*alertInfo)
+	if info.timer != nil {
+		info.timer.Stop()
+	}
+	return true
+}
+
+// processPendingAlert sends a "down" alert if the pending alert has expired and the system is still down.
+func (am *AlertManager) processPendingAlert(alertID string) {
+	value, loaded := am.pendingAlerts.LoadAndDelete(alertID)
+	if !loaded {
+		return
+	}
+
+	info := value.(*alertInfo)
+	if info.alertRecord.GetBool("triggered") {
+		return
+	}
+	if err := am.sendStatusAlert("down", info.systemName, info.alertRecord); err != nil {
+		am.hub.Logger().Error("Failed to send alert", "err", err)
+	}
+}
+
 // sendStatusAlert sends a status alert ("up" or "down") to the users associated with the alert records.
 func (am *AlertManager) sendStatusAlert(alertStatus string, systemName string, alertRecord *core.Record) error {
 	switch alertStatus {
@@ -177,8 +174,8 @@ func (am *AlertManager) sendStatusAlert(alertStatus string, systemName string, a
 	})
 }
 
-// resolveStatusAlerts resolves any status alerts that weren't resolved
-// when system came up (https://github.com/henrygd/beszel/issues/1052)
+// resolveStatusAlerts resolves any triggered status alerts that weren't resolved
+// when system came up (https://github.com/henrygd/beszel/issues/1052).
 func resolveStatusAlerts(app core.App) error {
 	db := app.DB()
 	// Find all active status alerts where the system is actually up
@@ -208,3 +205,36 @@ func resolveStatusAlerts(app core.App) error {
 	}
 	return nil
 }
+
+// restorePendingStatusAlerts re-queues untriggered status alerts for systems that
+// are still down after a hub restart. This rebuilds the lost in-memory timer state.
+func (am *AlertManager) restorePendingStatusAlerts() error {
+	type pendingStatusAlert struct {
+		AlertID    string `db:"alert_id"`
+		SystemName string `db:"system_name"`
+	}
+
+	var pending []pendingStatusAlert
+	err := am.hub.DB().NewQuery(`
+		SELECT a.id AS alert_id, s.name AS system_name
+		FROM alerts a
+		JOIN systems s ON a.system = s.id
+		WHERE a.name = 'Status'
+		AND a.triggered = false
+		AND s.status = 'down'
+	`).All(&pending)
+	if err != nil {
+		return err
+	}
+
+	for _, item := range pending {
+		alertRecord, err := am.hub.FindRecordById("alerts", item.AlertID)
+		if err != nil {
+			return err
+		}
+		min := max(1, alertRecord.GetInt("min"))
+		am.schedulePendingStatusAlert(item.SystemName, alertRecord, time.Duration(min)*time.Minute)
+	}
+
+	return nil
+}