internal/webhooks/webhook_service_health.go

Rama main · réplica de lectura del repositorio original.

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
}