package webhooks import ( "context" "fmt" "time" otelcodes "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace" svcerrors "github.com/sarathsp06/sparrow/internal/errors" "github.com/sarathsp06/sparrow/internal/storage" "github.com/sarathsp06/sparrow/internal/tenant" "github.com/sarathsp06/sparrow/internal/webhooks/store" ) // WebhookHealthData represents webhook health information type WebhookHealthData struct { WebhookID string `json:"webhook_id"` Health store.WebhookHealth `json:"health"` TotalDeliveries int `json:"total_deliveries"` SuccessfulDeliveries int `json:"successful_deliveries"` FailedDeliveries int `json:"failed_deliveries"` ConsecutiveFailures int `json:"consecutive_failures"` LastSuccessAt *time.Time `json:"last_success_at"` LastFailureAt *time.Time `json:"last_failure_at"` SuccessRate float64 `json:"success_rate"` AvgResponseTime int `json:"avg_response_time"` // milliseconds CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` // Error category breakdown ClientErrors int `json:"client_errors"` // 4xx responses ServerErrors int `json:"server_errors"` // 5xx responses TimeoutErrors int `json:"timeout_errors"` // Timeouts NetworkErrors int `json:"network_errors"` // DNS, TLS, connection refused, and other network errors UnexpectedStatusErrors int `json:"unexpected_status_errors"` // 2xx/3xx not in expected_status_codes } // HealthSummaryData represents health summary information type HealthSummaryData struct { HealthyCount int `json:"healthy_count"` DegradedCount int `json:"degraded_count"` UnhealthyCount int `json:"unhealthy_count"` UnknownCount int `json:"unknown_count"` TotalCount int `json:"total_count"` } // ConsumerStatsData represents consumer statistics type ConsumerStatsData struct { TotalWebhooks int `json:"total_webhooks"` ActiveWebhooks int `json:"active_webhooks"` TotalDeliveries int `json:"total_deliveries"` SuccessfulDeliveries int `json:"successful_deliveries"` FailedDeliveries int `json:"failed_deliveries"` PendingDeliveries int `json:"pending_deliveries"` SuccessRate float64 `json:"success_rate"` } // GetWebhookHealth retrieves health metrics for a webhook func (s *WebhookService) GetWebhookHealth(ctx context.Context, webhookID string, consumer string) (*WebhookHealthData, error) { span := trace.SpanFromContext(ctx) s.logger.InfoContext(ctx, "Processing get webhook health request", "webhook_id", webhookID, "consumer", consumer) // Validate required fields if webhookID == "" { return nil, svcerrors.Error(svcerrors.InvalidArgument, "webhook ID is required") } if consumer == "" { return nil, svcerrors.Error(svcerrors.InvalidArgument, "consumer is required") } tenantID := tenant.FromContext(ctx) id, err := parseUUID(webhookID, "webhook ID") if err != nil { return nil, err } // Get webhook to verify it exists and get current health webhook, err := s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer) if err != nil { span.RecordError(err) if storage.IsNotFound(err) { span.SetStatus(otelcodes.Error, "webhook not found") return nil, svcerrors.Error(svcerrors.NotFound, "webhook not found") } span.SetStatus(otelcodes.Error, "Failed to get webhook") s.logger.ErrorContext(ctx, "Failed to get webhook", "error", err) return nil, fmt.Errorf("failed to get webhook: %w", err) } if webhook == nil { return nil, svcerrors.Error(svcerrors.NotFound, "webhook not found") } // Get health state (current status and consecutive failures) healthState, err := s.webhookRepo.GetWebhookHealthState(ctx, tenantID, id) if err != nil { if !storage.IsNotFound(err) { span.RecordError(err) span.SetStatus(otelcodes.Error, "Failed to get webhook health state") s.logger.ErrorContext(ctx, "Failed to get webhook health state", "error", err) return nil, fmt.Errorf("failed to get webhook health state: %w", err) } // No health state yet: the health evaluator has not folded this // webhook's attempts (it runs every SPARROW_HEALTH_EVAL_INTERVAL). // The attempt counts below are read live from the attempt log, so // they are still reported; only the state fields stay empty. healthState = nil } // Metrics over the same window the health label is derived from. healthSummary, err := s.webhookRepo.GetWebhookHealthSummary(ctx, tenantID, id, store.DefaultHealthRules.WindowHours) if err != nil { s.logger.ErrorContext(ctx, "Failed to get health summary", "error", err) // Continue with just the state info } // Convert to response format healthData := &WebhookHealthData{ WebhookID: webhookID, Health: webhook.Health, } if healthState != nil { healthData.ConsecutiveFailures = healthState.ConsecutiveFailures healthData.LastSuccessAt = healthState.LastSuccessAt healthData.LastFailureAt = healthState.LastFailureAt healthData.CreatedAt = healthState.CreatedAt healthData.UpdatedAt = healthState.UpdatedAt } // Add summary data if available if healthSummary != nil { healthData.TotalDeliveries = healthSummary.TotalDeliveries healthData.SuccessfulDeliveries = healthSummary.SuccessfulDeliveries healthData.FailedDeliveries = healthSummary.FailedDeliveries healthData.SuccessRate = healthSummary.SuccessRate healthData.AvgResponseTime = healthSummary.AvgResponseTime healthData.ClientErrors = healthSummary.ClientErrors healthData.ServerErrors = healthSummary.ServerErrors healthData.TimeoutErrors = healthSummary.TimeoutErrors healthData.NetworkErrors = healthSummary.NetworkErrors healthData.UnexpectedStatusErrors = healthSummary.UnexpectedStatusErrors } s.logger.InfoContext(ctx, "Webhook health retrieved successfully", "webhook_id", webhookID, "health", webhook.Health, "success_rate", healthData.SuccessRate) return healthData, nil } // GetHealthSummary counts webhooks by health in one consumer, or across all // consumers when consumer is empty. func (s *WebhookService) GetHealthSummary(ctx context.Context, consumer string) (*HealthSummaryData, error) { span := trace.SpanFromContext(ctx) s.logger.InfoContext(ctx, "Processing get health summary request", "consumer", consumer) tenantID := tenant.FromContext(ctx) // Get health summary from repository summary, err := s.webhookRepo.GetHealthSummary(ctx, tenantID, consumer) if err != nil { span.RecordError(err) span.SetStatus(otelcodes.Error, "Failed to get health summary") s.logger.ErrorContext(ctx, "Failed to get health summary", "error", err) return nil, fmt.Errorf("failed to retrieve health summary: %w", err) } // Convert to response format healthSummary := &HealthSummaryData{ HealthyCount: summary[store.HealthHealthy], DegradedCount: summary[store.HealthDegraded], UnhealthyCount: summary[store.HealthUnhealthy], UnknownCount: summary[store.HealthUnknown], } // Calculate total healthSummary.TotalCount = healthSummary.HealthyCount + healthSummary.DegradedCount + healthSummary.UnhealthyCount + healthSummary.UnknownCount s.logger.InfoContext(ctx, "Health summary retrieved successfully", "healthy", healthSummary.HealthyCount, "degraded", healthSummary.DegradedCount, "unhealthy", healthSummary.UnhealthyCount, "unknown", healthSummary.UnknownCount, "total", healthSummary.TotalCount) return healthSummary, nil } // GetConsumerStats retrieves statistics for a consumer, or across all // consumers if empty. func (s *WebhookService) GetConsumerStats(ctx context.Context, consumer string) (*ConsumerStatsData, error) { s.logger.InfoContext(ctx, "Processing get consumer stats request", "consumer", consumer) tenantID := tenant.FromContext(ctx) stats, err := s.webhookRepo.GetConsumerStats(ctx, tenantID, consumer) if err != nil { s.logger.ErrorContext(ctx, "Failed to get consumer stats", "error", err) return nil, err } res := &ConsumerStatsData{ TotalWebhooks: stats.TotalWebhooks, ActiveWebhooks: stats.ActiveWebhooks, TotalDeliveries: stats.TotalDeliveries, SuccessfulDeliveries: stats.SuccessfulDeliveries, FailedDeliveries: stats.FailedDeliveries, PendingDeliveries: stats.PendingDeliveries, SuccessRate: stats.SuccessRate, } s.logger.InfoContext(ctx, "Consumer stats retrieved successfully", "consumer", consumer, "total_webhooks", res.TotalWebhooks, "active_webhooks", res.ActiveWebhooks, "success_rate", res.SuccessRate) return res, nil } // ListConsumers returns up to limit consumer names that own a webhook or an // event and contain query (case-insensitive), sorted by name. func (s *WebhookService) ListConsumers(ctx context.Context, query string, limit int32) ([]string, error) { span := trace.SpanFromContext(ctx) l, _ := normalizePagination(int(limit), 0) names, err := s.webhookRepo.ListConsumers(ctx, tenant.FromContext(ctx), query, l) if err != nil { span.RecordError(err) return nil, fmt.Errorf("failed to list consumers: %w", err) } return names, nil }