diff --git a/internal/api/alerts.go b/internal/api/alerts.go index 5bda3674d..2063ed604 100644 --- a/internal/api/alerts.go +++ b/internal/api/alerts.go @@ -13,20 +13,49 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/models" - "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/rcourtman/pulse-go-rewrite/internal/websocket" "github.com/rs/zerolog/log" ) +// AlertManager defines the interface for alert management operations. +type AlertManager interface { + GetConfig() alerts.AlertConfig + UpdateConfig(alerts.AlertConfig) + GetActiveAlerts() []alerts.Alert + NotifyExistingAlert(id string) + ClearAlertHistory() error + UnacknowledgeAlert(id string) error + AcknowledgeAlert(id, user string) error + ClearAlert(id string) bool + GetAlertHistory(limit int) []alerts.Alert + GetAlertHistorySince(since time.Time, limit int) []alerts.Alert +} + +// ConfigPersistence defines the interface for saving configuration. +type ConfigPersistence interface { + SaveAlertConfig(alerts.AlertConfig) error +} + +// AlertMonitor defines the interface for monitoring operations used by alert handlers. +type AlertMonitor interface { + GetAlertManager() AlertManager + GetConfigPersistence() ConfigPersistence + GetIncidentStore() *memory.IncidentStore + GetNotificationManager() *notifications.NotificationManager + SyncAlertState() + GetState() models.StateSnapshot +} + // AlertHandlers handles alert-related HTTP endpoints type AlertHandlers struct { - monitor *monitoring.Monitor + monitor AlertMonitor wsHub *websocket.Hub } // NewAlertHandlers creates new alert handlers -func NewAlertHandlers(monitor *monitoring.Monitor, wsHub *websocket.Hub) *AlertHandlers { +func NewAlertHandlers(monitor AlertMonitor, wsHub *websocket.Hub) *AlertHandlers { return &AlertHandlers{ monitor: monitor, wsHub: wsHub, @@ -34,7 +63,7 @@ func NewAlertHandlers(monitor *monitoring.Monitor, wsHub *websocket.Hub) *AlertH } // SetMonitor updates the monitor reference for alert handlers. -func (h *AlertHandlers) SetMonitor(m *monitoring.Monitor) { +func (h *AlertHandlers) SetMonitor(m AlertMonitor) { h.monitor = m } diff --git a/internal/api/audit_handlers.go b/internal/api/audit_handlers.go index d5b16e1e2..8405c9662 100644 --- a/internal/api/audit_handlers.go +++ b/internal/api/audit_handlers.go @@ -362,17 +362,17 @@ func (h *AuditHandlers) HandleExportAuditEvents(w http.ResponseWriter, r *http.R // Parse verification flag includeVerification := query.Get("verify") == "true" - // Get the SQLite logger + // Get the logger and check if it's persistent logger := audit.GetLogger() - sqliteLogger, ok := logger.(*audit.SQLiteLogger) + persistentLogger, ok := logger.(audit.PersistentLogger) if !ok { writeErrorResponse(w, http.StatusNotImplemented, "export_unavailable", - "Export requires SQLite audit logger", nil) + "Export requires persistent audit logger", nil) return } // Create exporter and export - exporter := audit.NewExporter(sqliteLogger) + exporter := audit.NewExporter(persistentLogger) result, err := exporter.Export(filter, format, includeVerification) if err != nil { writeErrorResponse(w, http.StatusInternalServerError, "export_failed", @@ -426,17 +426,17 @@ func (h *AuditHandlers) HandleAuditSummary(w http.ResponseWriter, r *http.Reques // Parse verification flag verifySignatures := query.Get("verify") == "true" - // Get the SQLite logger + // Get the logger and check if it's persistent logger := audit.GetLogger() - sqliteLogger, ok := logger.(*audit.SQLiteLogger) + persistentLogger, ok := logger.(audit.PersistentLogger) if !ok { writeErrorResponse(w, http.StatusNotImplemented, "summary_unavailable", - "Summary requires SQLite audit logger", nil) + "Summary requires persistent audit logger", nil) return } // Create exporter and generate summary - exporter := audit.NewExporter(sqliteLogger) + exporter := audit.NewExporter(persistentLogger) summary, err := exporter.GenerateSummary(filter, verifySignatures) if err != nil { writeErrorResponse(w, http.StatusInternalServerError, "summary_failed", diff --git a/internal/api/notifications.go b/internal/api/notifications.go index dd1cf8b44..31ae561dd 100644 --- a/internal/api/notifications.go +++ b/internal/api/notifications.go @@ -8,26 +8,61 @@ import ( "strings" "github.com/rcourtman/pulse-go-rewrite/internal/config" - "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/notifications" "github.com/rcourtman/pulse-go-rewrite/internal/utils" "github.com/rs/zerolog/log" ) +// NotificationManager defines the interface for notification management operations. +type NotificationManager interface { + GetEmailConfig() notifications.EmailConfig + SetEmailConfig(notifications.EmailConfig) + GetAppriseConfig() notifications.AppriseConfig + SetAppriseConfig(notifications.AppriseConfig) + GetWebhooks() []notifications.WebhookConfig + ValidateWebhookURL(string) error + AddWebhook(notifications.WebhookConfig) + UpdateWebhook(string, notifications.WebhookConfig) error + DeleteWebhook(string) error + SendTestWebhook(notifications.WebhookConfig) error + SendTestNotificationWithConfig(string, *notifications.EmailConfig, *notifications.TestNodeInfo) error + SendTestAppriseWithConfig(notifications.AppriseConfig) error + SendTestNotification(string) error + GetWebhookHistory() []notifications.WebhookDelivery + TestEnhancedWebhook(notifications.EnhancedWebhookConfig) (int, string, error) + GetQueueStats() (map[string]int, error) +} + +// NotificationConfigPersistence defines the interface for saving notification configuration. +type NotificationConfigPersistence interface { + SaveEmailConfig(notifications.EmailConfig) error + SaveAppriseConfig(notifications.AppriseConfig) error + SaveWebhooks([]notifications.WebhookConfig) error + IsEncryptionEnabled() bool +} + +// NotificationMonitor defines the interface for monitoring operations used by notification handlers. +type NotificationMonitor interface { + GetNotificationManager() NotificationManager + GetConfigPersistence() NotificationConfigPersistence + GetState() models.StateSnapshot +} + // NotificationHandlers handles notification-related HTTP endpoints type NotificationHandlers struct { - monitor *monitoring.Monitor + monitor NotificationMonitor } // NewNotificationHandlers creates new notification handlers -func NewNotificationHandlers(monitor *monitoring.Monitor) *NotificationHandlers { +func NewNotificationHandlers(monitor NotificationMonitor) *NotificationHandlers { return &NotificationHandlers{ monitor: monitor, } } // SetMonitor updates the monitor reference for notification handlers. -func (h *NotificationHandlers) SetMonitor(m *monitoring.Monitor) { +func (h *NotificationHandlers) SetMonitor(m NotificationMonitor) { h.monitor = m } @@ -697,25 +732,20 @@ func (h *NotificationHandlers) TestWebhook(w http.ResponseWriter, r *http.Reques func (h *NotificationHandlers) GetNotificationHealth(w http.ResponseWriter, r *http.Request) { // Get queue stats queueStats := make(map[string]interface{}) - if queue := h.monitor.GetNotificationManager().GetQueue(); queue != nil { - stats, err := queue.GetQueueStats() - if err != nil { - log.Warn().Err(err).Msg("Failed to get queue stats for health check") - queueStats["error"] = err.Error() - queueStats["healthy"] = false - } else { - queueStats = map[string]interface{}{ - "pending": stats["pending"], - "sending": stats["sending"], - "sent": stats["sent"], - "failed": stats["failed"], - "dlq": stats["dlq"], - "healthy": true, - } - } - } else { - queueStats["error"] = "queue not initialized" + stats, err := h.monitor.GetNotificationManager().GetQueueStats() + if err != nil { + log.Warn().Err(err).Msg("Failed to get queue stats for health check") + queueStats["error"] = err.Error() queueStats["healthy"] = false + } else { + queueStats = map[string]interface{}{ + "pending": stats["pending"], + "sending": stats["sending"], + "sent": stats["sent"], + "failed": stats["failed"], + "dlq": stats["dlq"], + "healthy": true, + } } // Get config status diff --git a/internal/api/profile_suggestions.go b/internal/api/profile_suggestions.go index 23956f606..809b0db21 100644 --- a/internal/api/profile_suggestions.go +++ b/internal/api/profile_suggestions.go @@ -8,7 +8,7 @@ import ( "strings" "time" - "github.com/rcourtman/pulse-go-rewrite/internal/ai/opencode" + "github.com/rcourtman/pulse-go-rewrite/internal/ai/chat" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rs/zerolog/log" @@ -115,7 +115,7 @@ Only include settings that are relevant to the user's request. Do not include se ctx, cancel := context.WithTimeout(r.Context(), 120*time.Second) defer cancel() - response, err := h.aiHandler.GetService().Execute(ctx, opencode.ExecuteRequest{ + response, err := h.aiHandler.GetService().Execute(ctx, chat.ExecuteRequest{ Prompt: fullPrompt, }) if err != nil { @@ -124,7 +124,7 @@ Only include settings that are relevant to the user's request. Do not include se return } - fullResponse := response.Message.Content + fullResponse, _ := response["content"].(string) if fullResponse == "" { log.Error().Msg("AI returned empty response") http.Error(w, "AI returned empty response", http.StatusInternalServerError) diff --git a/internal/api/updates.go b/internal/api/updates.go index 8f382d7be..a8ad3161f 100644 --- a/internal/api/updates.go +++ b/internal/api/updates.go @@ -17,15 +17,25 @@ import ( // UpdateHandlers handles update-related API requests type UpdateHandlers struct { - manager *updates.Manager + manager UpdateManager history *updates.UpdateHistory registry *updates.UpdaterRegistry statusRateLimits map[string]time.Time // IP -> last request time statusMu sync.RWMutex } +// UpdateManager defines the interface for update management operations +type UpdateManager interface { + CheckForUpdatesWithChannel(ctx context.Context, channel string) (*updates.UpdateInfo, error) + ApplyUpdate(ctx context.Context, req updates.ApplyUpdateRequest) error + GetStatus() updates.UpdateStatus + GetSSECachedStatus() (updates.UpdateStatus, time.Time) + AddSSEClient(w http.ResponseWriter, clientID string) *updates.SSEClient + RemoveSSEClient(clientID string) +} + // NewUpdateHandlers creates new update handlers -func NewUpdateHandlers(manager *updates.Manager, history *updates.UpdateHistory) *UpdateHandlers { +func NewUpdateHandlers(manager UpdateManager, history *updates.UpdateHistory) *UpdateHandlers { // Initialize updater registry registry := updates.NewUpdaterRegistry() @@ -142,7 +152,7 @@ func (h *UpdateHandlers) HandleUpdateStatus(w http.ResponseWriter, r *http.Reque h.statusMu.Unlock() // Get cached status from SSE broadcaster (more recent than manager status) - cachedStatus, cacheTime := h.manager.GetSSEBroadcaster().GetCachedStatus() + cachedStatus, cacheTime := h.manager.GetSSECachedStatus() // Add cache headers w.Header().Set("X-Cache", "HIT") @@ -193,8 +203,7 @@ func (h *UpdateHandlers) HandleUpdateStream(w http.ResponseWriter, r *http.Reque clientID := fmt.Sprintf("%s-%d", clientIP, time.Now().UnixNano()) // Register client with SSE broadcaster - broadcaster := h.manager.GetSSEBroadcaster() - client := broadcaster.AddClient(w, clientID) + client := h.manager.AddSSEClient(w, clientID) if client == nil { http.Error(w, "Streaming not supported", http.StatusInternalServerError) return @@ -222,7 +231,7 @@ func (h *UpdateHandlers) HandleUpdateStream(w http.ResponseWriter, r *http.Reque } // Clean up - broadcaster.RemoveClient(clientID) + h.manager.RemoveSSEClient(clientID) } // cleanupRateLimits periodically cleans up old entries from the rate limit map @@ -231,17 +240,19 @@ func (h *UpdateHandlers) cleanupRateLimits() { defer ticker.Stop() for range ticker.C { - now := time.Now() - h.statusMu.Lock() + h.doCleanupRateLimits(time.Now()) + } +} - // Remove entries older than 10 minutes - for ip, lastTime := range h.statusRateLimits { - if now.Sub(lastTime) > 10*time.Minute { - delete(h.statusRateLimits, ip) - } +func (h *UpdateHandlers) doCleanupRateLimits(now time.Time) { + h.statusMu.Lock() + defer h.statusMu.Unlock() + + // Remove entries older than 10 minutes + for ip, lastTime := range h.statusRateLimits { + if now.Sub(lastTime) > 10*time.Minute { + delete(h.statusRateLimits, ip) } - - h.statusMu.Unlock() } }