internal/webhooks/webhook_service.go
Rama main · réplica de lectura del repositorio original.
package webhooks
import (
"context"
"log/slog"
"math"
"net"
"regexp"
"time"
"github.com/google/uuid"
"github.com/sarathsp06/sparrow/internal/crypto"
svcerrors "github.com/sarathsp06/sparrow/internal/errors"
"github.com/sarathsp06/sparrow/internal/observability"
"github.com/sarathsp06/sparrow/internal/webhooks/client"
"github.com/sarathsp06/sparrow/internal/webhooks/queue"
"github.com/sarathsp06/sparrow/internal/webhooks/store"
)
type WebhookService struct {
jobInserter queue.JobInserter
webhookRepo store.RepositoryInterface
crypto *crypto.Service
logger *slog.Logger
metrics *observability.SparrowMetrics
networkPolicy client.NetworkPolicy
// autoRegisterEvents lets a push to an unknown event name create a
// schema-less event type instead of failing with NotFound.
autoRegisterEvents bool
// deliveryClient sends test events; nil turns them off.
deliveryClient *client.WebhookClient
testEvents testEventLimiter
}
// WebhookManager manages webhook registrations and their lifecycle.
type WebhookManager interface {
CreateWebhook(ctx context.Context, req WebhookRegistrationRequest) (reg *store.WebhookRegistration, secret string, err error)
UnregisterWebhook(ctx context.Context, webhookID string, consumer string) error
ListWebhooks(ctx context.Context, consumer string, webhookID string, event string, activeOnly bool, health string, limit, offset int32) ([]*store.WebhookRegistration, int32, error)
UpdateWebhookConfig(ctx context.Context, webhookID string, consumer string, events []string, url string, headers map[string]string, active bool, description string, httpConfig *WebhookHTTPConfig, secretHeaders map[string]string, signatureType string, requiresTransform bool, updateMask []string) error
RotateWebhookSecret(ctx context.Context, webhookID string, consumer string) (string, error)
PauseWebhook(ctx context.Context, webhookID string, consumer string, reason string) error
ResumeWebhook(ctx context.Context, webhookID string, consumer string) (*WebhookResumeResult, error)
GetConsumerStats(ctx context.Context, consumer string) (*ConsumerStatsData, error)
ListConsumers(ctx context.Context, query string, limit int32) ([]string, error)
}
// EventManager manages event type registrations and event publishing.
type EventManager interface {
RegisterEvent(ctx context.Context, name string, description string, schema map[string]any, metadata map[string]string, active bool) (string, time.Time, error)
ListEvents(ctx context.Context, activeOnly bool, consumer string, limit, offset int32) ([]*store.EventRegistration, int32, error)
UpdateEvent(ctx context.Context, name string, description string, schema map[string]any, metadata map[string]string, active bool, allowBreaking bool) (*EventTypeSaveResult, error)
GetEvent(ctx context.Context, name string) (*store.EventRegistration, error)
ListEventTypeVersions(ctx context.Context, name string) ([]*store.EventRegistrationVersion, error)
GetEventTypeVersion(ctx context.Context, name string, version int) (*store.EventRegistrationVersion, error)
ExportEventTypes(ctx context.Context, sel EventTypeExportSelection) ([]EventTypeDefinition, BundleStamp, error)
ImportEventTypes(ctx context.Context, items []EventTypeDefinition, stamp *BundleStamp, opts EventTypeImportOptions) (*EventTypeImportResult, error)
// PushEvent returns (eventID, isDuplicate, schemaValid, warnings, err).
// isDuplicate is true when idempotencyKey matched an existing event; the
// other fields then describe that existing event, not a new one.
PushEvent(ctx context.Context, consumer string, event string, payload map[string]any, ttlSeconds int64, metadata map[string]string, labels map[string]string, idempotencyKey *string, scheduledAt *time.Time) (eventID string, isDuplicate bool, schemaValid bool, warnings []string, err error)
RePushEvent(ctx context.Context, eventID string) (string, []string, error)
GetEventRecord(ctx context.Context, eventID string) (*store.EventRecord, int32, int32, int32, int32, error)
ListEventReports(ctx context.Context, filter store.EventReportFilter) (*EventReportPage, error)
}
// SubscriptionManager manages event subscriptions and payload-transform templates.
type SubscriptionManager interface {
CreateSubscription(ctx context.Context, webhookID, eventName, consumer string, headers map[string]string, method string, timeout int, transformEnabled bool, transformTemplate string, labelFilters map[string]string, settings SubscriptionTemplateSettings) (string, time.Time, error)
GetSubscription(ctx context.Context, subscriptionID string, consumer string) (*store.EventSubscription, error)
ListSubscriptions(ctx context.Context, consumer string, webhookID string, eventName string, limit, offset int32) ([]*store.EventSubscription, int32, error)
UpdateSubscription(ctx context.Context, subscriptionID string, consumer string, headers map[string]string, method string, timeout int, transformEnabled bool, transformTemplate string, labelFilters map[string]string, settings SubscriptionTemplateSettings, meta TemplateSaveMeta) error
// ListSubscriptionTemplateVersions returns the saved template history of a
// subscription, newest (current) first.
ListSubscriptionTemplateVersions(ctx context.Context, subscriptionID string, consumer string, limit int) ([]*store.SubscriptionTemplateVersion, error)
DeleteSubscription(ctx context.Context, subscriptionID string, consumer string) error
PauseSubscription(ctx context.Context, subscriptionID, consumer, reason string) (*store.EventSubscription, error)
ResumeSubscription(ctx context.Context, subscriptionID, consumer string) (*ResumeResult, error)
TestSubscriptionTemplate(ctx context.Context, eventName, transformTemplate, consumer string, strict bool, payload map[string]any) (string, error)
ListSubscriptionsByWebhookIDs(ctx context.Context, webhookIDs []uuid.UUID) ([]*store.EventSubscription, error)
GetTemplateFunctions() []TemplateFunctionInfo
}
// DeliveryManager exposes delivery status, attempt history, and retries.
type DeliveryManager interface {
GetDeliveryStatus(ctx context.Context, deliveryID string, consumer string) (*store.WebhookDelivery, error)
GetDeliveryAttempts(ctx context.Context, deliveryID string, consumer string) ([]*store.WebhookHealthEvent, error)
ListDeliveries(ctx context.Context, filter store.DeliveryFilter) (*DeliveryPage, error)
RetryDelivery(ctx context.Context, consumer string, deliveryID string, webhookID string, force bool) ([]string, int32, error)
}
// HealthManager exposes per-webhook and aggregate health.
type HealthManager interface {
GetWebhookHealth(ctx context.Context, webhookID string, consumer string) (*WebhookHealthData, error)
GetHealthSummary(ctx context.Context, consumer string) (*HealthSummaryData, error)
}
// BatchManager runs prepared bulk re-push and bulk retry jobs.
type BatchManager interface {
RePushEvents(ctx context.Context, scope BatchScope, repushID string) error
GetRepushStatus(ctx context.Context, scope BatchScope, repushID string) (*store.BatchJob, error)
CancelRepush(ctx context.Context, scope BatchScope, repushID string) error
RetryDeliveries(ctx context.Context, scope BatchScope, retryID string) error
GetRetryStatus(ctx context.Context, scope BatchScope, retryID string) (*store.BatchJob, error)
CancelRetry(ctx context.Context, scope BatchScope, retryID string) error
}
// AlertConfigManager manages opt-in email recipients for Sparrow's
// self-generated webhook.health_changed / webhook.delivery_failed events.
type AlertConfigManager interface {
CreateAlertConfig(ctx context.Context, consumer, webhookID, email string, eventTypes []string) (*store.AlertConfig, error)
ListAlertConfigs(ctx context.Context, consumer string, webhookID string) ([]*store.AlertConfig, error)
DeleteAlertConfig(ctx context.Context, consumer, id string) error
AlertDeliveryConfigured(ctx context.Context) (bool, error)
}
// SecretRevealer decrypts stored webhook secrets for masking and exposes the
// Ed25519 signing public key. Used by response conversion, never for delivery.
type SecretRevealer interface {
DecryptSecretHeaders(encrypted []byte) (map[string]string, error)
DecryptWebhookSecret(encrypted []byte) (string, error)
WebhookSigningPublicKeyHex(encryptedPrivKey []byte) string
}
//go:generate go tool gowrap gen -g -i WebhookServiceInterface -t ../../templates/opentelemetry.tmpl -o WebhookServiceInterface_otel.go
// WebhookServiceInterface is the composite of every domain interface, satisfied
// by *WebhookService. Consumers should depend on the narrow per-domain
// interfaces; the composite exists for wiring and the OTel decorator.
type WebhookServiceInterface interface {
WebhookManager
EventManager
SubscriptionManager
DeliveryManager
HealthManager
BatchManager
SecretRevealer
AlertConfigManager
TestEventSender
}
type TemplateFunctionInfo struct {
Name string
Description string
}
var _ WebhookServiceInterface = (*WebhookService)(nil)
// NewWebhookService creates a new WebhookService instance
// WebhookServiceOption configures a WebhookService.
type WebhookServiceOption func(*WebhookService)
// WithAllowPrivateNetworks disables SSRF protection for webhook URL validation,
// permitting loopback and private-network addresses. Useful for self-hosted
// deployments where webhook targets live on the same network, and required for
// integration tests that use httptest.NewServer.
func WithAllowPrivateNetworks(allow bool) WebhookServiceOption {
return func(s *WebhookService) {
s.networkPolicy.AllowPrivate = allow
}
}
// WithAllowedNetworks permits webhook URLs that resolve into these CIDRs
// (e.g. internal services on a VPN) while every other private, loopback and
// cloud-metadata address stays blocked.
func WithAllowedNetworks(nets []*net.IPNet) WebhookServiceOption {
return func(s *WebhookService) {
s.networkPolicy.AllowedNetworks = nets
}
}
// WithAutoRegisterEvents makes a push to an unregistered event name create a
// schema-less event type instead of returning NotFound. Off by default: it is
// a development convenience, and in production it turns typos into permanent
// event type names, since event types are never deleted.
func WithAutoRegisterEvents(enabled bool) WebhookServiceOption {
return func(s *WebhookService) {
s.autoRegisterEvents = enabled
}
}
func NewWebhookService(queueManager queue.JobInserter, webhookRepo store.RepositoryInterface, cryptoSvc *crypto.Service, opts ...WebhookServiceOption) *WebhookService {
metrics, err := observability.NewSparrowMetrics()
if err != nil {
// Log error but continue without metrics
slog.Default().With("component", "webhook-service").Error("Failed to initialize metrics", "error", err)
}
svc := &WebhookService{
jobInserter: queueManager,
webhookRepo: webhookRepo,
crypto: cryptoSvc,
logger: slog.Default().With("component", "webhook-service"),
metrics: metrics,
}
for _, opt := range opts {
opt(svc)
}
return svc
}
// --- Common Helpers ---
// parseUUID parses a UUID string and returns a typed error on failure.
// entityName is used in the error message (e.g. "webhook ID", "subscription ID").
func parseUUID(s string, entityName string) (uuid.UUID, error) {
id, err := uuid.Parse(s)
if err != nil {
return uuid.Nil, svcerrors.Errorf(svcerrors.InvalidArgument, "invalid %s: %v", entityName, err)
}
return id, nil
}
// maxPageLimit caps list page sizes to bound response size and DB load.
const maxPageLimit = 1000
// normalizePagination applies default limit (50), caps limit at maxPageLimit,
// and ensures offset is non-negative.
// DeliveryPage is one page of a delivery list. RetryID and RetryTotal are
// set when the list was asked to snapshot its matches for a batch retry.
type DeliveryPage struct {
Items []*store.WebhookDelivery
HasMore bool
RetryID string
RetryTotal int
}
// EventReportPage is one page of an event occurrence list. RepushID and
// RepushTotal are set when the list was asked to snapshot its matches for a
// batch re-push.
type EventReportPage struct {
Items []*store.EventReportWithStats
HasMore bool
RepushID string
RepushTotal int
}
func normalizePagination(limit, offset int) (int, int) {
if limit <= 0 {
limit = 50
} else if limit > maxPageLimit {
limit = maxPageLimit
}
if offset < 0 {
offset = 0
}
return limit, offset
}
// clampInt32 converts n for the int32 API fields, saturating instead of
// wrapping when n is out of range.
func clampInt32(n int) int32 {
return int32(max(math.MinInt32, min(n, math.MaxInt32))) //nolint:gosec // clamped to the int32 range above
}
// Label validation constraints
const (
maxLabelsPerMap = 20 // max key-value pairs per labels/labelFilters map
maxLabelKeyLen = 64 // max characters per label key
maxLabelValueLen = 256 // max characters per label value
)
// labelKeyPattern restricts label keys to alphanumeric, dot, underscore, and hyphen.
var labelKeyPattern = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`)
// validateLabels checks that a labels/label_filters map meets size and format constraints.
func validateLabels(m map[string]string, fieldName string) error {
if len(m) > maxLabelsPerMap {
return svcerrors.Errorf(svcerrors.InvalidArgument, "%s: too many entries (%d), maximum is %d", fieldName, len(m), maxLabelsPerMap)
}
for k, v := range m {
if k == "" {
return svcerrors.Errorf(svcerrors.InvalidArgument, "%s: key must not be empty", fieldName)
}
if len(k) > maxLabelKeyLen {
return svcerrors.Errorf(svcerrors.InvalidArgument, "%s: key %q exceeds maximum length of %d characters", fieldName, k, maxLabelKeyLen)
}
if !labelKeyPattern.MatchString(k) {
return svcerrors.Errorf(svcerrors.InvalidArgument, "%s: key %q contains invalid characters (allowed: alphanumeric, '.', '_', '-')", fieldName, k)
}
if len(v) > maxLabelValueLen {
return svcerrors.Errorf(svcerrors.InvalidArgument, "%s: value for key %q exceeds maximum length of %d characters", fieldName, k, maxLabelValueLen)
}
}
return nil
}