Fix critical notification system bugs and security issues

Critical fixes (P0):
- Fix cooldown timing: Mark cooldown only after successful delivery, not before enqueue
- Add os.MkdirAll to queue initialization to prevent silent failures on fresh installs
- Add DNS re-validation at webhook send time to prevent DNS rebinding SSRF attacks
- Add SSRF validation for Apprise HTTP URLs
- Remove secret logging (bot tokens, routing keys) from debug logs
- Implement lastNotified cleanup to prevent unbounded memory growth
- Use shared HTTP client for webhooks to enable TLS connection reuse
- Add fallback to direct sending when queue enqueue fails
- Make queue worker concurrent (5 workers with semaphore) to prevent head-of-line blocking
- Fix webhook rate limiter race condition with separate mutex
- Fix email manager thread safety with mutex on rate limiter
- Fix grouping timer leak by adding stopCleanup signal
- Fix webhook 429 double sleep (use Retry-After OR backoff, not both)

Frontend improvements:
- Add queue/DLQ management API methods (getQueueStats, getDLQ, retryDLQItem, deleteDLQItem)
- Add getNotificationHealth and getWebhookHistory endpoints
- Add Apprise test support to NotificationTestRequest type

Related to notification system audit

frontend-modern/src/api/notifications.ts

@@ -71,7 +71,7 @@ export interface AppriseConfig {
 }
 
 export interface NotificationTestRequest {
-  type: 'email' | 'webhook';
+  type: 'email' | 'webhook' | 'apprise';
   config?: Record<string, unknown>; // Backend expects different format than frontend types
   webhookId?: string;
 }
@@ -197,4 +197,53 @@ export class NotificationsAPI {
       body: JSON.stringify(webhook),
     });
   }
+
+  // Queue and DLQ management
+  static async getQueueStats(): Promise<Record<string, number>> {
+    return apiFetchJSON(`${this.baseUrl}/queue/stats`);
+  }
+
+  static async getDLQ(limit = 100): Promise<unknown[]> {
+    return apiFetchJSON(`${this.baseUrl}/dlq?limit=${limit}`);
+  }
+
+  static async retryDLQItem(id: string): Promise<{ success: boolean }> {
+    return apiFetchJSON(`${this.baseUrl}/dlq/retry`, {
+      method: 'POST',
+      body: JSON.stringify({ id }),
+    });
+  }
+
+  static async deleteDLQItem(id: string): Promise<{ success: boolean }> {
+    return apiFetchJSON(`${this.baseUrl}/dlq/delete`, {
+      method: 'POST',
+      body: JSON.stringify({ id }),
+    });
+  }
+
+  static async getNotificationHealth(): Promise<{
+    queue?: {
+      healthy: boolean;
+      pending: number;
+      sending: number;
+      sent: number;
+      failed: number;
+      dlq: number;
+    };
+    email: {
+      enabled: boolean;
+      configured: boolean;
+    };
+    webhooks: {
+      total: number;
+      enabled: number;
+    };
+    healthy: boolean;
+  }> {
+    return apiFetchJSON(`${this.baseUrl}/health`);
+  }
+
+  static async getWebhookHistory(): Promise<unknown[]> {
+    return apiFetchJSON(`${this.baseUrl}/webhook-history`);
+  }
 }

internal/notifications/email_enhanced.go

@@ -20,6 +20,7 @@ type EnhancedEmailManager struct {
 
 // RateLimiter implements a simple rate limiter
 type RateLimiter struct {
+	mu        sync.Mutex
 	rate      int
 	lastSent  time.Time
 	sentCount int
@@ -83,6 +84,9 @@ func (e *EnhancedEmailManager) checkRateLimit() error {
 		return nil // No rate limit
 	}
 
+	e.rateLimit.mu.Lock()
+	defer e.rateLimit.mu.Unlock()
+
 	now := time.Now()
 	if now.Sub(e.rateLimit.lastSent) >= time.Minute {
 		// Reset counter after a minute

internal/notifications/notifications.go

@@ -121,8 +121,11 @@ type NotificationManager struct {
 	groupByGuest      bool
 	webhookHistory    []WebhookDelivery            // Keep last 100 webhook deliveries for debugging
 	webhookRateLimits map[string]*webhookRateLimit // Track rate limits per webhook URL
+	webhookRateMu     sync.Mutex                   // Separate mutex for webhook rate limiting
 	appriseExec       appriseExecFunc
 	queue             *NotificationQueue // Persistent notification queue
+	webhookClient     *http.Client       // Shared HTTP client for webhooks
+	stopCleanup       chan struct{}      // Signal to stop cleanup goroutine
 }
 
 type appriseExecFunc func(ctx context.Context, path string, args []string) ([]byte, error)
@@ -350,6 +353,8 @@ func NewNotificationManager(publicURL string) *NotificationManager {
 		publicURL:         cleanURL,
 		appriseExec:       defaultAppriseExec,
 		queue:             queue,
+		webhookClient:     createSecureWebhookClient(WebhookTimeout),
+		stopCleanup:       make(chan struct{}),
 	}
 
 	// Wire up queue processor if queue is available
@@ -357,6 +362,9 @@ func NewNotificationManager(publicURL string) *NotificationManager {
 		queue.SetProcessor(nm.ProcessQueuedNotification)
 	}
 
+	// Start periodic cleanup of old lastNotified entries (every 1 hour)
+	go nm.cleanupOldNotificationRecords()
+
 	return nm
 }
 
@@ -660,7 +668,10 @@ func (n *NotificationManager) sendGroupedAlerts() {
 }
 
 // enqueueNotifications adds notifications to the persistent queue
+// Falls back to direct sending if enqueue fails
 func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webhooks []WebhookConfig, appriseConfig AppriseConfig, alertsToSend []*alerts.Alert) {
+	anyFailed := false
+
 	// Enqueue email notification
 	if emailConfig.Enabled {
 		configJSON, err := json.Marshal(emailConfig)
@@ -674,7 +685,9 @@ func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webh
 				MaxAttempts: 3,
 			}
 			if err := n.queue.Enqueue(notif); err != nil {
-				log.Error().Err(err).Msg("Failed to enqueue email notification")
+				log.Error().Err(err).Msg("Failed to enqueue email notification - falling back to direct send")
+				anyFailed = true
+				go n.sendGroupedEmail(emailConfig, alertsToSend)
 			} else {
 				log.Debug().Int("alertCount", len(alertsToSend)).Msg("Enqueued email notification")
 			}
@@ -695,7 +708,9 @@ func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webh
 					MaxAttempts: 3,
 				}
 				if err := n.queue.Enqueue(notif); err != nil {
-					log.Error().Err(err).Str("webhookName", webhook.Name).Msg("Failed to enqueue webhook notification")
+					log.Error().Err(err).Str("webhookName", webhook.Name).Msg("Failed to enqueue webhook notification - falling back to direct send")
+					anyFailed = true
+					go n.sendGroupedWebhook(webhook, alertsToSend)
 				} else {
 					log.Debug().Str("webhookName", webhook.Name).Int("alertCount", len(alertsToSend)).Msg("Enqueued webhook notification")
 				}
@@ -716,12 +731,27 @@ func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webh
 				MaxAttempts: 3,
 			}
 			if err := n.queue.Enqueue(notif); err != nil {
-				log.Error().Err(err).Msg("Failed to enqueue apprise notification")
+				log.Error().Err(err).Msg("Failed to enqueue apprise notification - falling back to direct send")
+				anyFailed = true
+				go n.sendGroupedApprise(appriseConfig, alertsToSend)
 			} else {
 				log.Debug().Int("alertCount", len(alertsToSend)).Msg("Enqueued apprise notification")
 			}
 		}
 	}
+
+	// If any enqueue failed, mark cooldown immediately for fire-and-forget sends
+	if anyFailed {
+		n.mu.Lock()
+		now := time.Now()
+		for _, alert := range alertsToSend {
+			n.lastNotified[alert.ID] = notificationRecord{
+				lastSent:   now,
+				alertStart: alert.StartTime,
+			}
+		}
+		n.mu.Unlock()
+	}
 }
 
 // sendNotificationsDirect sends notifications without using the queue (fallback)
@@ -1367,8 +1397,8 @@ func (n *NotificationManager) sendGroupedWebhook(webhook WebhookConfig, alertLis
 
 // checkWebhookRateLimit checks if a webhook can be sent based on rate limits
 func (n *NotificationManager) checkWebhookRateLimit(webhookURL string) bool {
-	n.mu.Lock()
+	n.webhookRateMu.Lock()
-	defer n.mu.Unlock()
+	defer n.webhookRateMu.Unlock()
 
 	now := time.Now()
 	limit, exists := n.webhookRateLimits[webhookURL]
@@ -1485,10 +1515,8 @@ func (n *NotificationManager) sendWebhookRequest(webhook WebhookConfig, jsonData
 			Msg("Sending webhook")
 	}
 
-	// Send request with secure client
+	// Send request with shared secure client
-	client := createSecureWebhookClient(WebhookTimeout)
+	resp, err := n.webhookClient.Do(req)
-
-	resp, err := client.Do(req)
 	if err != nil {
 		log.Error().
 			Err(err).
@@ -2370,11 +2398,49 @@ func (n *NotificationManager) ProcessQueuedNotification(notif *QueuedNotificatio
 	return err
 }
 
+// cleanupOldNotificationRecords periodically cleans up old entries from lastNotified map
+func (n *NotificationManager) cleanupOldNotificationRecords() {
+	ticker := time.NewTicker(1 * time.Hour)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			n.mu.Lock()
+			now := time.Now()
+			cutoff := now.Add(-24 * time.Hour)
+			cleaned := 0
+
+			for alertID, record := range n.lastNotified {
+				// Remove entries older than 24 hours
+				if record.lastSent.Before(cutoff) {
+					delete(n.lastNotified, alertID)
+					cleaned++
+				}
+			}
+
+			if cleaned > 0 {
+				log.Debug().
+					Int("cleaned", cleaned).
+					Int("remaining", len(n.lastNotified)).
+					Msg("Cleaned up old notification cooldown records")
+			}
+			n.mu.Unlock()
+		case <-n.stopCleanup:
+			// Stop cleanup when manager is stopped
+			return
+		}
+	}
+}
+
 // Stop gracefully stops the notification manager
 func (n *NotificationManager) Stop() {
 	n.mu.Lock()
 	defer n.mu.Unlock()
 
+	// Stop cleanup goroutine
+	close(n.stopCleanup)
+
 	// Stop the notification queue if it exists
 	if n.queue != nil {
 		n.queue.Stop()

internal/notifications/queue.go

@@ -55,6 +55,7 @@ type NotificationQueue struct {
 	cleanupTicker   *time.Ticker
 	notifyChan      chan struct{}                    // Signal when new notifications are added
 	processor       func(*QueuedNotification) error  // Notification processor function
+	workerSem       chan struct{}                    // Semaphore for limiting concurrent workers
 }
 
 // NewNotificationQueue creates a new persistent notification queue
@@ -98,6 +99,7 @@ func NewNotificationQueue(dataDir string) (*NotificationQueue, error) {
 		processorTicker: time.NewTicker(5 * time.Second),
 		cleanupTicker:   time.NewTicker(1 * time.Hour),
 		notifyChan:      make(chan struct{}, 100),
+		workerSem:       make(chan struct{}, 5), // Allow 5 concurrent workers
 	}
 
 	if err := nq.initSchema(); err != nil {
@@ -520,9 +522,9 @@ func (nq *NotificationQueue) SetProcessor(processor func(*QueuedNotification) er
 	nq.processor = processor
 }
 
-// processBatch processes a batch of pending notifications
+// processBatch processes a batch of pending notifications concurrently
 func (nq *NotificationQueue) processBatch() {
-	pending, err := nq.GetPending(10)
+	pending, err := nq.GetPending(20) // Increased batch size for concurrency
 	if err != nil {
 		log.Error().Err(err).Msg("Failed to get pending notifications")
 		return
@@ -534,9 +536,21 @@ func (nq *NotificationQueue) processBatch() {
 
 	log.Debug().Int("count", len(pending)).Msg("Processing notification batch")
 
+	// Process notifications concurrently with semaphore limiting
+	var wg sync.WaitGroup
 	for _, notif := range pending {
-		nq.processNotification(notif)
+		wg.Add(1)
+		go func(n *QueuedNotification) {
+			defer wg.Done()
+
+			// Acquire semaphore slot
+			nq.workerSem <- struct{}{}
+			defer func() { <-nq.workerSem }()
+
+			nq.processNotification(n)
+		}(notif)
 	}
+	wg.Wait()
 }
 
 // processNotification processes a single notification

internal/notifications/webhook_enhanced.go

@@ -207,15 +207,7 @@ func (n *NotificationManager) sendWebhookWithRetry(webhook EnhancedWebhookConfig
 
 	for attempt := 0; attempt <= maxRetries; attempt++ {
 		if attempt > 0 {
-			log.Debug().
+			// Check for Retry-After header from previous response (overrides backoff)
-				Str("webhook", webhook.Name).
-				Int("attempt", attempt).
-				Int("maxRetries", maxRetries).
-				Dur("backoff", backoff).
-				Msg("Retrying webhook after backoff")
-			time.Sleep(backoff)
-
-			// Check for Retry-After header from previous response
 			if lastResp != nil && lastResp.StatusCode == 429 {
 				if retryAfter := lastResp.Header.Get("Retry-After"); retryAfter != "" {
 					if seconds, err := strconv.Atoi(retryAfter); err == nil {
@@ -225,12 +217,30 @@ func (n *NotificationManager) sendWebhookWithRetry(webhook EnhancedWebhookConfig
 							Dur("retryAfter", customBackoff).
 							Msg("Using Retry-After header for backoff")
 						time.Sleep(customBackoff)
-						backoff = customBackoff
+						backoff = customBackoff // Use this for next iteration
 					}
+				} else {
+					// No Retry-After, use exponential backoff
+					log.Debug().
+						Str("webhook", webhook.Name).
+						Int("attempt", attempt).
+						Int("maxRetries", maxRetries).
+						Dur("backoff", backoff).
+						Msg("Retrying webhook after backoff")
+					time.Sleep(backoff)
 				}
+			} else {
+				// Not a 429, use exponential backoff
+				log.Debug().
+					Str("webhook", webhook.Name).
+					Int("attempt", attempt).
+					Int("maxRetries", maxRetries).
+					Dur("backoff", backoff).
+					Msg("Retrying webhook after backoff")
+				time.Sleep(backoff)
 			}
 
-			// Exponential backoff
+			// Exponential backoff for next iteration
 			backoff *= 2
 			if backoff > WebhookMaxBackoff {
 				backoff = WebhookMaxBackoff