Files
pulse/internal/api/alerting/notification_queue.go
T
2026-08-23 09:05:54 +01:00

248 lines
7.8 KiB
Go

package alerting
import (
"encoding/json"
"net/http"
"strconv"
"github.com/rcourtman/pulse-go-rewrite/internal/api/apihttp"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
"github.com/rcourtman/pulse-go-rewrite/internal/notifications"
"github.com/rcourtman/pulse-go-rewrite/internal/utils"
"github.com/rs/zerolog/log"
)
// NotificationQueueHandlers handles notification queue API endpoints
type NotificationQueueHandlers struct {
monitor *monitoring.Monitor
}
// NewNotificationQueueHandlers creates new notification queue handlers
func NewNotificationQueueHandlers(monitor *monitoring.Monitor) *NotificationQueueHandlers {
return &NotificationQueueHandlers{
monitor: monitor,
}
}
// GetDLQ returns notifications in the dead letter queue
func (h *NotificationQueueHandlers) GetDLQ(w http.ResponseWriter, r *http.Request) {
if !apihttp.EnsureScope(w, r, config.ScopeMonitoringRead) {
return
}
limit := 100
if limitStr := r.URL.Query().Get("limit"); limitStr != "" {
if l, err := strconv.Atoi(limitStr); err == nil && l > 0 && l <= 1000 {
limit = l
}
}
queue := h.monitor.GetNotificationManager().GetQueue()
if queue == nil {
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
return
}
dlq, err := queue.GetDLQ(limit)
if err != nil {
log.Error().Err(err).Msg("Failed to get DLQ")
http.Error(w, "Failed to retrieve dead letter queue", http.StatusInternalServerError)
return
}
if err := utils.WriteJSONResponse(w, dlq); err != nil {
log.Error().Err(err).Msg("Failed to write DLQ response")
}
}
// GetQueueStats returns statistics about the notification queue
func (h *NotificationQueueHandlers) GetQueueStats(w http.ResponseWriter, r *http.Request) {
if !apihttp.EnsureScope(w, r, config.ScopeMonitoringRead) {
return
}
queue := h.monitor.GetNotificationManager().GetQueue()
if queue == nil {
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
return
}
stats, err := queue.GetQueueStats()
if err != nil {
log.Error().Err(err).Msg("Failed to get queue stats")
http.Error(w, "Failed to retrieve queue statistics", http.StatusInternalServerError)
return
}
if err := utils.WriteJSONResponse(w, stats); err != nil {
log.Error().Err(err).Msg("Failed to write queue stats response")
}
}
// RetryDLQItem retries a specific notification from the DLQ
func (h *NotificationQueueHandlers) RetryDLQItem(w http.ResponseWriter, r *http.Request) {
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
return
}
// Limit request body to 8KB to prevent memory exhaustion
r.Body = http.MaxBytesReader(w, r.Body, 8*1024)
var request struct {
ID string `json:"id"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
http.Error(w, "Invalid request body", http.StatusBadRequest)
return
}
if request.ID == "" {
http.Error(w, "Missing notification ID", http.StatusBadRequest)
return
}
queue := h.monitor.GetNotificationManager().GetQueue()
if queue == nil {
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
return
}
// Reset notification to pending status with immediate retry
if err := queue.ScheduleRetry(request.ID, 0); err != nil {
log.Error().Err(err).Str("id", request.ID).Msg("Failed to retry DLQ item")
http.Error(w, "Failed to retry notification", http.StatusInternalServerError)
return
}
log.Info().Str("id", request.ID).Msg("DLQ notification scheduled for retry")
if err := utils.WriteJSONResponse(w, map[string]interface{}{
"success": true,
"message": "Notification scheduled for retry",
"id": request.ID,
}); err != nil {
log.Error().Err(err).Msg("Failed to write retry response")
}
}
// DeleteDLQItem removes a notification from the DLQ permanently
func (h *NotificationQueueHandlers) DeleteDLQItem(w http.ResponseWriter, r *http.Request) {
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
return
}
// Limit request body to 8KB to prevent memory exhaustion
r.Body = http.MaxBytesReader(w, r.Body, 8*1024)
var request struct {
ID string `json:"id"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
http.Error(w, "Invalid request body", http.StatusBadRequest)
return
}
if request.ID == "" {
http.Error(w, "Missing notification ID", http.StatusBadRequest)
return
}
queue := h.monitor.GetNotificationManager().GetQueue()
if queue == nil {
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
return
}
// Update status to deleted/cancelled
if err := queue.UpdateStatus(request.ID, notifications.QueueStatusCancelled, "Manually deleted from DLQ"); err != nil {
log.Error().Err(err).Str("id", request.ID).Msg("Failed to delete DLQ item")
http.Error(w, "Failed to delete notification", http.StatusInternalServerError)
return
}
log.Info().Str("id", request.ID).Msg("DLQ notification deleted")
if err := utils.WriteJSONResponse(w, map[string]interface{}{
"success": true,
"message": "Notification deleted from DLQ",
"id": request.ID,
}); err != nil {
log.Error().Err(err).Msg("Failed to write delete response")
}
}
// RetryTerminalFailures returns all retained terminal failures to the queue
// after an operator has repaired the destination.
func (h *NotificationQueueHandlers) RetryTerminalFailures(w http.ResponseWriter, r *http.Request) {
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
return
}
queue := h.monitor.GetNotificationManager().GetQueue()
if queue == nil {
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
return
}
affected, err := queue.RetryTerminalFailures()
if err != nil {
log.Error().Err(err).Msg("Failed to retry retained terminal notifications")
http.Error(w, "Failed to retry retained notification failures", http.StatusInternalServerError)
return
}
if err := utils.WriteJSONResponse(w, map[string]interface{}{
"success": true,
"affected": affected,
}); err != nil {
log.Error().Err(err).Msg("Failed to write terminal retry response")
}
}
// DismissTerminalFailures cancels all retained terminal queue rows without
// deleting their immutable delivery audit.
func (h *NotificationQueueHandlers) DismissTerminalFailures(w http.ResponseWriter, r *http.Request) {
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
return
}
queue := h.monitor.GetNotificationManager().GetQueue()
if queue == nil {
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
return
}
affected, err := queue.DismissTerminalFailures()
if err != nil {
log.Error().Err(err).Msg("Failed to dismiss retained terminal notifications")
http.Error(w, "Failed to dismiss retained notification failures", http.StatusInternalServerError)
return
}
if err := utils.WriteJSONResponse(w, map[string]interface{}{
"success": true,
"affected": affected,
}); err != nil {
log.Error().Err(err).Msg("Failed to write terminal dismissal response")
}
}
// HandleNotificationQueue routes notification queue requests
func (h *NotificationQueueHandlers) HandleNotificationQueue(w http.ResponseWriter, r *http.Request) {
path := r.URL.Path
switch {
case path == "/api/notifications/dlq" && r.Method == http.MethodGet:
h.GetDLQ(w, r)
case path == "/api/notifications/queue/stats" && r.Method == http.MethodGet:
h.GetQueueStats(w, r)
case path == "/api/notifications/dlq/retry" && r.Method == http.MethodPost:
h.RetryDLQItem(w, r)
case path == "/api/notifications/dlq/delete" && r.Method == http.MethodPost:
h.DeleteDLQItem(w, r)
case path == "/api/notifications/terminal-failures/retry" && r.Method == http.MethodPost:
h.RetryTerminalFailures(w, r)
case path == "/api/notifications/terminal-failures/dismiss" && r.Method == http.MethodPost:
h.DismissTerminalFailures(w, r)
default:
http.Error(w, "Not found", http.StatusNotFound)
}
}