Fix mixed quiet-hours notification replay queueing

internal/notifications/notifications.go

@@ -73,6 +73,11 @@ type notificationDeliveryJob struct {
 	AppriseConfig *AppriseConfig
 }
 
+type notificationQueueBucket struct {
+	job         notificationDeliveryJob
+	nextRetryAt *time.Time
+}
+
 // createSecureWebhookClient creates an HTTP client with security controls
 func (n *NotificationManager) createSecureWebhookClient(timeout time.Duration) *http.Client {
 	return n.createSecureWebhookClientWithTLS(timeout, false)
@@ -342,35 +347,50 @@ func annotateResolvedMetadata(alert *alerts.Alert, resolvedAt time.Time) {
 	alert.Metadata[metadataResolvedAt] = resolvedAt.Format(time.RFC3339)
 }
 
+func quietHoursReplayAtForAlert(alert *alerts.Alert, now time.Time) *time.Time {
+	if alert == nil || alert.Metadata == nil {
+		return nil
+	}
+	raw, ok := alert.Metadata[alerts.MetadataQuietHoursReplayAt]
+	if !ok {
+		return nil
+	}
+
+	var parsed time.Time
+	switch value := raw.(type) {
+	case string:
+		ts, err := time.Parse(time.RFC3339, strings.TrimSpace(value))
+		if err != nil {
+			return nil
+		}
+		parsed = ts
+	case time.Time:
+		parsed = value
+	case float64:
+		if value <= 0 {
+			return nil
+		}
+		parsed = time.Unix(int64(value), 0)
+	default:
+		return nil
+	}
+
+	if !parsed.After(now) {
+		return nil
+	}
+	parsed = parsed.UTC()
+	return &parsed
+}
+
 func quietHoursReplayAtForAlerts(alertList []*alerts.Alert, now time.Time) *time.Time {
 	var replayAt time.Time
 	for _, alert := range alertList {
-		if alert == nil || alert.Metadata == nil {
+		alertReplayAt := quietHoursReplayAtForAlert(alert, now)
+		if alertReplayAt == nil {
 			continue
 		}
-		raw, ok := alert.Metadata[alerts.MetadataQuietHoursReplayAt]
-		if !ok {
-			continue
-		}
-
-		var parsed time.Time
-		switch value := raw.(type) {
-		case string:
-			ts, err := time.Parse(time.RFC3339, strings.TrimSpace(value))
-			if err != nil {
-				continue
-			}
-			parsed = ts
-		case time.Time:
-			parsed = value
-		case float64:
-			if value <= 0 {
-				continue
-			}
-			parsed = time.Unix(int64(value), 0)
-		}
-		if parsed.After(now) && (replayAt.IsZero() || parsed.After(replayAt)) {
-			replayAt = parsed
+		if replayAt.IsZero() || alertReplayAt.After(replayAt) {
+			replayAt = *alertReplayAt
 		}
 	}
 
@@ -381,6 +401,39 @@ func quietHoursReplayAtForAlerts(alertList []*alerts.Alert, now time.Time) *time
 	return &replayAt
 }
 
+func notificationQueueBucketsForJob(job notificationDeliveryJob, now time.Time) []notificationQueueBucket {
+	if len(job.Alerts) == 0 {
+		return []notificationQueueBucket{{job: job}}
+	}
+
+	buckets := make([]notificationQueueBucket, 0, len(job.Alerts))
+	bucketIndexes := make(map[string]int, len(job.Alerts))
+	for _, alert := range job.Alerts {
+		nextRetryAt := quietHoursReplayAtForAlert(alert, now)
+		key := "immediate"
+		if nextRetryAt != nil {
+			key = nextRetryAt.Format(time.RFC3339Nano)
+		}
+
+		index, exists := bucketIndexes[key]
+		if !exists {
+			bucketJob := job
+			bucketJob.Alerts = nil
+			bucket := notificationQueueBucket{job: bucketJob}
+			if nextRetryAt != nil {
+				replayAt := *nextRetryAt
+				bucket.nextRetryAt = &replayAt
+			}
+			index = len(buckets)
+			bucketIndexes[key] = index
+			buckets = append(buckets, bucket)
+		}
+		buckets[index].job.Alerts = append(buckets[index].job.Alerts, alert)
+	}
+
+	return buckets
+}
+
 // NormalizeAppriseConfig cleans and normalizes Apprise configuration values.
 func NormalizeAppriseConfig(cfg AppriseConfig) AppriseConfig {
 	normalized := cfg
@@ -1182,31 +1235,33 @@ func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue,
 			continue
 		}
 
-		notif := &QueuedNotification{
-			Type:        queueTypeForNotificationDeliveryJob(job),
-			Alerts:      job.Alerts,
-			Config:      configJSON,
-			MaxAttempts: 3,
-			NextRetryAt: quietHoursReplayAtForAlerts(job.Alerts, time.Now()),
-		}
-		if err := queue.Enqueue(notif); err != nil {
-			anyFailed = true
-			n.logNotificationJobError(job, err, "failed to enqueue notification - falling back to direct send")
-			n.dispatchNotificationJobAsync(job, "failed to send notification after queue enqueue failure")
-			continue
-		}
+		for _, bucket := range notificationQueueBucketsForJob(job, time.Now()) {
+			notif := &QueuedNotification{
+				Type:        queueTypeForNotificationDeliveryJob(bucket.job),
+				Alerts:      bucket.job.Alerts,
+				Config:      configJSON,
+				MaxAttempts: 3,
+				NextRetryAt: bucket.nextRetryAt,
+			}
+			if err := queue.Enqueue(notif); err != nil {
+				anyFailed = true
+				n.logNotificationJobError(bucket.job, err, "failed to enqueue notification - falling back to direct send")
+				n.dispatchNotificationJobAsync(bucket.job, "failed to send notification after queue enqueue failure")
+				continue
+			}
 
-		logger := log.Debug().
-			Str("type", job.Type).
-			Str("event", string(job.Event)).
-			Int("alertCount", len(job.Alerts))
-		if notif.NextRetryAt != nil {
-			logger = logger.Time("nextRetryAt", *notif.NextRetryAt)
+			logger := log.Debug().
+				Str("type", bucket.job.Type).
+				Str("event", string(bucket.job.Event)).
+				Int("alertCount", len(bucket.job.Alerts))
+			if notif.NextRetryAt != nil {
+				logger = logger.Time("nextRetryAt", *notif.NextRetryAt)
+			}
+			if bucket.job.WebhookConfig != nil {
+				logger = logger.Str("webhookName", bucket.job.WebhookConfig.Name)
+			}
+			logger.Msg("enqueued notification delivery job")
 		}
-		if job.WebhookConfig != nil {
-			logger = logger.Str("webhookName", job.WebhookConfig.Name)
-		}
-		logger.Msg("enqueued notification delivery job")
 	}
 
 	if anyFailed && len(jobs) > 0 && jobs[0].Event == eventResolved {

internal/notifications/notifications_test.go

@@ -2,6 +2,7 @@ package notifications
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"io"
 	"net/http"
@@ -3087,6 +3088,105 @@ func TestProcessQueuedNotification_QuietHoursReplayUsesQueueRetryTime(t *testing
 	}
 }
 
+func TestEnqueueNotificationJobs_MixedQuietHoursGroupKeepsImmediateAlertsReady(t *testing.T) {
+	queue, err := NewNotificationQueue(t.TempDir())
+	if err != nil {
+		t.Fatalf("failed to create queue: %v", err)
+	}
+	defer func() { _ = queue.Stop() }()
+
+	replayAt := time.Now().Add(time.Hour).UTC()
+	immediate := &alerts.Alert{
+		ID:           "immediate-critical",
+		Type:         "cpu",
+		Level:        alerts.AlertLevelCritical,
+		ResourceName: "node-1",
+		Message:      "critical alert should send now",
+		StartTime:    time.Now().Add(-time.Minute),
+	}
+	deferred := &alerts.Alert{
+		ID:           "quiet-hours-warning",
+		Type:         "memory",
+		Level:        alerts.AlertLevelWarning,
+		ResourceName: "node-2",
+		Message:      "warning alert should wait",
+		StartTime:    time.Now().Add(-time.Minute),
+		Metadata: map[string]interface{}{
+			alerts.MetadataQuietHoursSuppressed:        true,
+			alerts.MetadataQuietHoursSuppressionReason: "non-critical",
+			alerts.MetadataQuietHoursReplayAt:          replayAt.Format(time.RFC3339),
+		},
+	}
+	jobs := buildNotificationDeliveryJobs(
+		EmailConfig{Enabled: true, SMTPHost: "smtp.example.com", SMTPPort: 587, From: "alerts@example.com", To: []string{"ops@example.com"}},
+		nil,
+		AppriseConfig{},
+		[]*alerts.Alert{immediate, deferred},
+		eventAlert,
+		time.Time{},
+	)
+
+	nm := &NotificationManager{}
+	if failed := nm.enqueueNotificationJobs(queue, jobs); failed {
+		t.Fatal("did not expect mixed quiet-hours enqueue to fall back")
+	}
+
+	pending, err := queue.GetPending(10)
+	if err != nil {
+		t.Fatalf("GetPending failed: %v", err)
+	}
+	if len(pending) != 1 {
+		t.Fatalf("expected one immediately pending notification, got %d", len(pending))
+	}
+	if pending[0].NextRetryAt != nil {
+		t.Fatalf("immediate notification next retry = %v, want nil", pending[0].NextRetryAt)
+	}
+	if len(pending[0].Alerts) != 1 || pending[0].Alerts[0].ID != immediate.ID {
+		t.Fatalf("pending alerts = %#v, want only %q", pending[0].Alerts, immediate.ID)
+	}
+
+	rows, err := queue.db.Query(`SELECT alerts, next_retry_at FROM notification_queue WHERE type = ? ORDER BY next_retry_at IS NOT NULL, next_retry_at`, "email")
+	if err != nil {
+		t.Fatalf("failed to read queued notifications: %v", err)
+	}
+	defer rows.Close()
+
+	type queuedRow struct {
+		alerts      []*alerts.Alert
+		nextRetryAt sql.NullInt64
+	}
+	var queued []queuedRow
+	for rows.Next() {
+		var row queuedRow
+		var alertsJSON string
+		if err := rows.Scan(&alertsJSON, &row.nextRetryAt); err != nil {
+			t.Fatalf("failed to scan queued notification: %v", err)
+		}
+		if err := json.Unmarshal([]byte(alertsJSON), &row.alerts); err != nil {
+			t.Fatalf("failed to unmarshal queued alerts: %v", err)
+		}
+		queued = append(queued, row)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatalf("iterate queued notifications: %v", err)
+	}
+	if len(queued) != 2 {
+		t.Fatalf("queued rows = %d, want 2", len(queued))
+	}
+	if queued[0].nextRetryAt.Valid {
+		t.Fatalf("immediate row next_retry_at = %d, want NULL", queued[0].nextRetryAt.Int64)
+	}
+	if len(queued[0].alerts) != 1 || queued[0].alerts[0].ID != immediate.ID {
+		t.Fatalf("immediate row alerts = %#v, want only %q", queued[0].alerts, immediate.ID)
+	}
+	if !queued[1].nextRetryAt.Valid || queued[1].nextRetryAt.Int64 != replayAt.Unix() {
+		t.Fatalf("deferred row next_retry_at = %#v, want %d", queued[1].nextRetryAt, replayAt.Unix())
+	}
+	if len(queued[1].alerts) != 1 || queued[1].alerts[0].ID != deferred.ID {
+		t.Fatalf("deferred row alerts = %#v, want only %q", queued[1].alerts, deferred.ID)
+	}
+}
+
 func TestGetQueue_NilQueue(t *testing.T) {
 	nm := &NotificationManager{}