internal/webhooks/webhook_service_registration.go

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

package webhooks

import (
	"context"
	"fmt"
	neturl "net/url"
	"slices"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/lib/pq"
	"go.opentelemetry.io/otel/attribute"
	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/client"
	"github.com/sarathsp06/sparrow/internal/webhooks/store"
)

// setWebhookActive is the shared implementation of PauseWebhook and ResumeWebhook.
// It loads the webhook, checks whether a state transition is needed, and persists the change.
func (s *WebhookService) setWebhookActive(ctx context.Context, webhookID string, consumer string, active bool) error {
	if webhookID == "" {
		return svcerrors.Error(svcerrors.InvalidArgument, "webhook ID is required")
	}
	if consumer == "" {
		return svcerrors.Error(svcerrors.InvalidArgument, "consumer is required")
	}

	tenantID := tenant.FromContext(ctx)

	id, err := parseUUID(webhookID, "webhook ID")
	if err != nil {
		return err
	}

	webhook, err := s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer)
	if err != nil {
		s.logger.ErrorContext(ctx, "Failed to get webhook", "error", err)
		return fmt.Errorf("failed to retrieve webhook: %w", err)
	}
	if webhook.Active == active {
		if active {
			return svcerrors.Error(svcerrors.FailedPrecondition, "webhook is already active")
		}
		return svcerrors.Error(svcerrors.FailedPrecondition, "webhook is already paused")
	}
	webhook.Active = active
	webhook.UpdatedAt = time.Now()
	if err := s.webhookRepo.UpdateWebhook(ctx, tenantID, webhook); err != nil {
		action := "pause"
		if active {
			action = "resume"
		}
		s.logger.ErrorContext(ctx, "Failed to "+action+" webhook", "error", err)
		return fmt.Errorf("failed to %s webhook: %w", action, err)
	}

	action := "paused"
	if active {
		action = "resumed"
	}
	s.logger.InfoContext(ctx, "Webhook "+action+" successfully", "webhook_id", webhookID)
	return nil
}

// CreateWebhook creates a webhook registration with HTTP configuration support.
// It returns the registration as stored and the plaintext signing secret,
// which callers show once: the stored registration only has it encrypted.
func (s *WebhookService) CreateWebhook(ctx context.Context, req WebhookRegistrationRequest) (*store.WebhookRegistration, string, error) {
	redactedURL := client.RedactURL(req.URL)
	span := trace.SpanFromContext(ctx)
	span.SetAttributes(
		attribute.String("consumer", req.Consumer),
		attribute.StringSlice("events", req.Events),
		attribute.String("url", redactedURL),
	)

	s.logger.InfoContext(ctx, "Processing enhanced webhook creation request",
		"consumer", req.Consumer,
		"events", req.Events,
		"url", redactedURL,
	)

	tenantID := tenant.FromContext(ctx)

	if req.Consumer == "" {
		return nil, "", svcerrors.Error(svcerrors.InvalidArgument, "consumer is required")
	}

	// Validate webhook URL against SSRF
	if err := ValidateWebhookURL(req.URL, s.networkPolicy); err != nil {
		return nil, "", err
	}
	if err := validateHeaders("headers", stringHeaders(req.Headers)); err != nil {
		return nil, "", err
	}
	secretHeaders, err := canonicalHeaders("secret_headers", req.SecretHeaders)
	if err != nil {
		return nil, "", err
	}
	if err := validateHeaders("secret_headers", secretHeaders); err != nil {
		return nil, "", err
	}
	req.SecretHeaders = secretHeaders

	if err := req.TemplateSettings.validate(); err != nil {
		return nil, "", err
	}
	if err := validateTransformTemplate(req.TransformTemplate); err != nil {
		return nil, "", err
	}
	if req.RequiresTransform && len(req.Events) > 0 && strings.TrimSpace(req.TransformTemplate) == "" {
		return nil, "", svcerrors.Error(svcerrors.InvalidArgument, "requires_transform is set, so transform_template is required for the subscriptions created from events (or register without events and add each subscription with its own template)")
	}

	// Convert request to internal webhook registration
	webhookReg, err := req.ToWebhookRegistration()
	if err != nil {
		return nil, "", svcerrors.Wrapf(err, svcerrors.InvalidArgument, "invalid webhook configuration: %v", err)
	}

	// Generate ID if not provided
	if webhookReg.ID == "" {
		webhookReg.ID = uuid.New().String()
	}

	// Validate event names
	for _, event := range req.Events {
		if event == "" {
			return nil, "", svcerrors.Error(svcerrors.InvalidArgument, "empty event name not allowed")
		}
		if err := checkEventForConsumer(req.Consumer, event); err != nil {
			return nil, "", err
		}
	}

	webhookID, err := parseUUID(webhookReg.ID, "webhook ID")
	if err != nil {
		return nil, "", err
	}

	// Convert internal webhook to store model for database operation
	storeWebhook := &store.WebhookRegistration{
		ID:                    webhookID,
		Consumer:              webhookReg.Consumer,
		URL:                   webhookReg.URL,
		Timeout:               webhookReg.HTTPConfig.RequestTimeoutSeconds,
		Active:                webhookReg.Active,
		Description:           webhookReg.Description,
		Health:                store.WebhookHealth(webhookReg.Health),
		MaxRetries:            DerefIntOr(webhookReg.HTTPConfig.MaxRetries, DefaultMaxRetries),
		RetryBackoffSeconds:   webhookReg.HTTPConfig.RetryBackoffSeconds,
		CaptureResponseBody:   DerefBoolOr(webhookReg.HTTPConfig.CaptureResponseBody, false),
		FollowRedirects:       DerefBoolOr(webhookReg.HTTPConfig.FollowRedirects, true),
		VerifySSL:             DerefBoolOr(webhookReg.HTTPConfig.VerifySSL, true),
		RequestTimeoutSeconds: webhookReg.HTTPConfig.RequestTimeoutSeconds,
		UserAgent:             webhookReg.HTTPConfig.UserAgent,
		ContentType:           webhookReg.HTTPConfig.ContentType,
		RateLimitRPS:          webhookReg.HTTPConfig.RateLimitRPS,
		RequiresTransform:     req.RequiresTransform,
		CreatedAt:             time.Now(),
		UpdatedAt:             time.Now(),
	}

	// Set signature type, defaulting to HMAC
	sigType := store.SignatureType(webhookReg.SignatureType)
	if sigType != store.SignatureTypeEd25519 {
		sigType = store.SignatureTypeHMAC
	}
	storeWebhook.SignatureType = sigType

	// Convert headers to string map for store model
	headersMap := make(map[string]string)
	for k, v := range webhookReg.Headers {
		if str, ok := v.(string); ok {
			headersMap[k] = str
		}
	}
	storeWebhook.Headers = headersMap

	// Encrypt webhook secret if provided, or generate one if missing
	if webhookReg.HTTPConfig.WebhookSecret != "" {
		encSecret, err := s.EncryptWebhookSecret(webhookReg.HTTPConfig.WebhookSecret)
		if err != nil {
			return nil, "", fmt.Errorf("failed to encrypt webhook secret: %w", err)
		}
		storeWebhook.WebhookSecret = encSecret
	} else {
		// Auto-generate a secret if missing to ensure all webhooks are signed
		secret, err := generateWebhookSecret()
		if err != nil {
			return nil, "", fmt.Errorf("failed to generate webhook secret: %w", err)
		}
		encSecret, err := s.EncryptWebhookSecret(secret)
		if err != nil {
			return nil, "", fmt.Errorf("failed to encrypt generated webhook secret: %w", err)
		}
		storeWebhook.WebhookSecret = encSecret
		webhookReg.HTTPConfig.WebhookSecret = secret // returned in plaintext once
	}

	// Generate Ed25519 keypair only when signature_type is "ed25519".
	// The private key is envelope-encrypted and stored; the public key is derived at runtime.
	if sigType == store.SignatureTypeEd25519 {
		encPrivKey, err := s.generateEncryptedEd25519Key()
		if err != nil {
			return nil, "", err
		}
		storeWebhook.Ed25519PrivateKey = encPrivKey
	}

	// Encrypt secret headers if provided
	if len(req.SecretHeaders) > 0 {
		encrypted, err := s.EncryptSecretHeaders(req.SecretHeaders)
		if err != nil {
			return nil, "", fmt.Errorf("failed to encrypt secret headers: %w", err)
		}
		storeWebhook.SecretHeaders = encrypted
	}

	// Convert expected status codes
	expectedCodes := make(pq.Int64Array, len(webhookReg.HTTPConfig.ExpectedStatusCodes))
	for i, code := range webhookReg.HTTPConfig.ExpectedStatusCodes {
		expectedCodes[i] = int64(code)
	}
	storeWebhook.ExpectedStatusCodes = expectedCodes

	// Build subscriptions slice for atomic creation
	var subscriptions []*store.EventSubscription
	for _, event := range req.Events {
		subscriptions = append(subscriptions, &store.EventSubscription{
			EventName:          event,
			Consumer:           req.Consumer,
			TransformEnabled:   strings.TrimSpace(req.TransformTemplate) != "",
			TransformTemplate:  req.TransformTemplate,
			OnTransformError:   req.TemplateSettings.OnTransformError,
			TemplateMissingKey: req.TemplateSettings.TemplateMissingKey,
		})
	}

	// Atomically register the webhook and all subscriptions in a single transaction
	firstVersion := store.SubscriptionTemplateVersion{Source: req.TemplateMeta.Source, Notes: req.TemplateMeta.Notes, SavedBy: req.TemplateMeta.SavedBy}
	if err := s.webhookRepo.RegisterWebhookWithSubscriptions(ctx, tenantID, storeWebhook, subscriptions, firstVersion); err != nil {
		s.logger.ErrorContext(ctx, "Failed to register webhook",
			"consumer", req.Consumer,
			"events", req.Events,
			"url", client.RedactURL(req.URL),
			"error", err,
		)
		span.RecordError(err)
		span.SetStatus(otelcodes.Error, "failed to register webhook")
		return nil, "", fmt.Errorf("failed to register webhook: %w", err)
	}

	// Initialize rate limit state if rate limiting is configured
	if storeWebhook.RateLimitRPS != nil {
		if err := s.webhookRepo.UpsertRateLimitState(ctx, tenantID, webhookID); err != nil {
			s.logger.WarnContext(ctx, "Failed to initialize rate limit state",
				"webhook_id", webhookReg.ID,
				"error", err,
			)
			// Non-fatal: the rate limit row will be created on first delivery slot acquisition
		}
	}

	// Update metrics
	if s.metrics != nil {
		s.metrics.WebhookRegistrations.Add(ctx, 1)
	}

	s.logger.InfoContext(ctx, "Enhanced webhook registered successfully",
		"webhook_id", webhookReg.ID,
		"consumer", req.Consumer,
		"events", req.Events,
		"url", client.RedactURL(req.URL),
		"http_config_provided", req.HTTPConfig != nil,
	)

	span.SetStatus(otelcodes.Ok, "webhook created successfully")

	return storeWebhook, webhookReg.HTTPConfig.WebhookSecret, nil
}

// UnregisterWebhook removes a webhook registration
func (s *WebhookService) UnregisterWebhook(ctx context.Context, webhookID string, consumer string) error {
	s.logger.InfoContext(ctx, "Processing webhook un registration request",
		"webhook_id", webhookID,
		"consumer", consumer,
	)
	if webhookID == "" {
		return svcerrors.Error(svcerrors.InvalidArgument, "webhook_id is required")
	}
	if consumer == "" {
		return svcerrors.Error(svcerrors.InvalidArgument, "consumer is required")
	}

	tenantID := tenant.FromContext(ctx)

	id, err := parseUUID(webhookID, "webhook ID")
	if err != nil {
		return err
	}

	// Check if webhook exists in consumer
	_, err = s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer)
	if err != nil {
		return fmt.Errorf("failed to retrieve webhook: %w", err)
	}

	if err := s.webhookRepo.UnregisterWebhook(ctx, tenantID, id); err != nil {
		s.logger.ErrorContext(ctx, "Failed to unregister webhook",
			"webhook_id", webhookID,
			"error", err,
		)
		return fmt.Errorf("failed to unregister webhook: %w", err)
	}
	s.logger.InfoContext(ctx, "Webhook unregistered successfully",
		"webhook_id", webhookID,
	)
	return nil
}

// ListWebhooks lists all registered webhooks with optional consumer and other filters.
// When consumer is empty, returns webhooks across all consumers.
// When health is non-empty, only webhooks with that health status are returned.
func (s *WebhookService) ListWebhooks(ctx context.Context, consumer string, webhookID string, event string, activeOnly bool, health string, limit, offset int32) ([]*store.WebhookRegistration, int32, error) {
	s.logger.InfoContext(ctx, "Processing list webhooks request",
		"consumer", consumer,
		"webhook_id", webhookID,
		"event", event,
		"active_only", activeOnly,
		"limit", limit,
		"offset", offset,
	)

	tenantID := tenant.FromContext(ctx)

	l, o := normalizePagination(int(limit), int(offset))
	limit, offset = clampInt32(l), clampInt32(o)

	if webhookID != "" {
		id, err := parseUUID(webhookID, "webhook ID")
		if err != nil {
			return nil, 0, err
		}

		reg, err := s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer)
		if err != nil {
			if storage.IsNotFound(err) {
				return []*store.WebhookRegistration{}, 0, nil
			}
			return nil, 0, fmt.Errorf("failed to retrieve webhook: %w", err)
		}
		if activeOnly && !reg.Active {
			return []*store.WebhookRegistration{}, 0, nil
		}
		if health != "" && string(reg.Health) != health {
			return []*store.WebhookRegistration{}, 0, nil
		}
		return []*store.WebhookRegistration{reg}, 1, nil
	}

	registrations, totalCount, err := s.webhookRepo.ListWebhooksPaginated(ctx, tenantID, consumer, event, activeOnly, store.WebhookHealth(health), int(limit), int(offset))
	if err != nil {
		s.logger.ErrorContext(ctx, "Failed to list webhooks",
			"consumer", consumer,
			"error", err,
		)
		return nil, 0, fmt.Errorf("failed to list webhooks: %w", err)
	}

	s.logger.InfoContext(ctx, "Listed webhooks successfully",
		"consumer", consumer,
		"count", len(registrations),
		"total", totalCount,
	)
	return registrations, clampInt32(totalCount), nil
}

// PauseWebhook temporarily disables webhook deliveries
func (s *WebhookService) PauseWebhook(ctx context.Context, webhookID string, consumer string, reason string) error {
	s.logger.InfoContext(ctx, "Pausing webhook", "webhook_id", webhookID, "consumer", consumer, "reason", reason)
	return s.setWebhookActive(ctx, webhookID, consumer, false)
}

// WebhookResumeResult describes a resumed webhook.
type WebhookResumeResult struct {
	// PausedDeliveries counts the webhook's deliveries held as paused. They
	// are not sent automatically; the caller decides what to retry.
	PausedDeliveries int
}

// ResumeWebhook re-enables webhook deliveries, whether it was paused by an
// operator or auto-disabled. Deliveries held while it was paused stay paused
// until retried, like those of a resumed subscription.
func (s *WebhookService) ResumeWebhook(ctx context.Context, webhookID string, consumer string) (*WebhookResumeResult, error) {
	s.logger.InfoContext(ctx, "Resuming webhook", "webhook_id", webhookID, "consumer", consumer)
	if err := s.setWebhookActive(ctx, webhookID, consumer, true); err != nil {
		return nil, err
	}
	id, err := parseUUID(webhookID, "webhook ID")
	if err != nil {
		return nil, err
	}
	status := string(store.StatusPaused)
	count, err := s.webhookRepo.CountDeliveries(ctx, tenant.FromContext(ctx), store.DeliveryFilter{
		Consumer:  consumer,
		WebhookID: &id,
		Status:    &status,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to count paused deliveries: %w", err)
	}
	return &WebhookResumeResult{PausedDeliveries: count}, nil
}

// UpdateWebhookConfig updates webhook configuration.
// When updateMask is non-empty, only the listed field paths are applied.
// When updateMask is empty, falls back to legacy behavior (all non-zero fields applied).
//
// Supported mask paths:
//
//	"url", "active", "description", "events", "headers", "secret_headers",
//	"signature_type", "http_config", "http_config.webhook_secret",
//	"http_config.rate_limit_rps", "requires_transform"
//
// requires_transform is only changed when named in the mask. Turning it on
// fails while any subscription lacks a transform, and a webhook that requires
// one cannot have its events replaced in bulk (that recreates subscriptions
// without templates).
func (s *WebhookService) 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 {
	s.logger.InfoContext(ctx, "Processing update webhook config request",
		"webhook_id", webhookID,
		"consumer", consumer,
		"update_mask", updateMask)

	if webhookID == "" {
		return svcerrors.Error(svcerrors.InvalidArgument, "webhook ID is required")
	}
	if consumer == "" {
		return svcerrors.Error(svcerrors.InvalidArgument, "consumer is required")
	}

	tenantID := tenant.FromContext(ctx)

	webhookUUID, err := parseUUID(webhookID, "webhook ID")
	if err != nil {
		return err
	}

	webhook, err := s.webhookRepo.GetWebhookByID(ctx, tenantID, webhookUUID, consumer)
	if err != nil {
		s.logger.ErrorContext(ctx, "Failed to get webhook", "error", err)
		return fmt.Errorf("failed to retrieve webhook: %w", err)
	}

	oldURL := webhook.URL
	oldSecretHeaders := webhook.SecretHeaders

	// Build a set for O(1) lookup. When empty, all non-zero fields are applied (legacy).
	mask := make(map[string]bool, len(updateMask))
	for _, p := range updateMask {
		mask[p] = true
	}
	useMask := len(mask) > 0

	shouldUpdate := func(field string) bool {
		if !useMask {
			return true // legacy: apply everything
		}
		return mask[field]
	}

	// Build new subscription list (pure logic, no DB calls)
	var newSubs []*store.EventSubscription
	replaceEvents := shouldUpdate("events") && len(events) > 0
	if replaceEvents {
		for _, event := range events {
			if err := checkEventForConsumer(consumer, event); err != nil {
				return err
			}
			newSubs = append(newSubs, &store.EventSubscription{
				EventName: event,
			})
		}
	}

	// Apply config mutations to the in-memory webhook (pure logic, no DB calls)
	if shouldUpdate("url") && url != "" {
		normalizedURL := strings.TrimSpace(url)
		if normalizedURL == "" {
			return svcerrors.Error(svcerrors.InvalidArgument, "URL is required")
		}
		if err := ValidateWebhookURL(normalizedURL, s.networkPolicy); err != nil {
			return err
		}
		webhook.URL = normalizedURL
	}
	if shouldUpdate("headers") && headers != nil {
		if err := validateHeaders("headers", headers); err != nil {
			return err
		}
		webhook.Headers = headers
	}
	if shouldUpdate("active") {
		webhook.Active = active
	}
	if shouldUpdate("description") && description != "" {
		webhook.Description = description
	}
	// Apply HTTP config updates if provided
	if shouldUpdate("http_config") && httpConfig != nil {
		if httpConfig.MaxRetries != nil {
			webhook.MaxRetries = *httpConfig.MaxRetries
		}
		if httpConfig.RetryBackoffSeconds > 0 {
			webhook.RetryBackoffSeconds = httpConfig.RetryBackoffSeconds
		}
		if httpConfig.RequestTimeoutSeconds > 0 {
			webhook.RequestTimeoutSeconds = httpConfig.RequestTimeoutSeconds
		}
		if len(httpConfig.ExpectedStatusCodes) > 0 {
			int64Codes := make([]int64, len(httpConfig.ExpectedStatusCodes))
			for i, c := range httpConfig.ExpectedStatusCodes {
				int64Codes[i] = int64(c)
			}
			webhook.ExpectedStatusCodes = int64Codes
		}
		// Only update the webhook secret if explicitly requested via mask.
		// Without mask (legacy mode), non-empty secret is applied.
		// With mask, "http_config.webhook_secret" must be in the mask.
		updateSecret := false
		if useMask {
			updateSecret = mask["http_config.webhook_secret"]
		} else {
			updateSecret = httpConfig.WebhookSecret != ""
		}
		if updateSecret && httpConfig.WebhookSecret != "" {
			encSecret, err := s.EncryptWebhookSecret(httpConfig.WebhookSecret)
			if err != nil {
				return fmt.Errorf("failed to encrypt webhook secret: %w", err)
			}
			webhook.WebhookSecret = encSecret
		}
		if httpConfig.UserAgent != "" {
			webhook.UserAgent = httpConfig.UserAgent
		}
		if httpConfig.ContentType != "" {
			webhook.ContentType = httpConfig.ContentType
		}
		if httpConfig.CaptureResponseBody != nil {
			webhook.CaptureResponseBody = *httpConfig.CaptureResponseBody
		}
		if httpConfig.FollowRedirects != nil {
			webhook.FollowRedirects = *httpConfig.FollowRedirects
		}
		if httpConfig.VerifySSL != nil {
			webhook.VerifySSL = *httpConfig.VerifySSL
		}
		// RateLimitRPS: pointer field — nil means "not provided" (keep existing),
		// non-nil overrides. With mask, "http_config.rate_limit_rps" must be in the mask.
		updateRateLimit := false
		if useMask {
			updateRateLimit = mask["http_config.rate_limit_rps"]
		} else {
			updateRateLimit = httpConfig.RateLimitRPS != nil
		}
		if updateRateLimit {
			webhook.RateLimitRPS = httpConfig.RateLimitRPS
		}
		// Enforce the same bounds the create path enforces: validate the
		// merged result, not just the delta (e.g. request_timeout_seconds
		// must stay within 1-300 after a PATCH).
		codes := make(IntArray, len(webhook.ExpectedStatusCodes))
		for i, code := range webhook.ExpectedStatusCodes {
			codes[i] = int(code)
		}
		merged := WebhookHTTPConfig{
			MaxRetries:            &webhook.MaxRetries,
			RetryBackoffSeconds:   webhook.RetryBackoffSeconds,
			RequestTimeoutSeconds: webhook.RequestTimeoutSeconds,
			ExpectedStatusCodes:   codes,
			ContentType:           webhook.ContentType,
			RateLimitRPS:          webhook.RateLimitRPS,
		}
		if err := merged.ValidateConfig(); err != nil {
			return err
		}
	}
	// Merge secret header updates into the existing encrypted map.
	// Empty-string values remove a header; omitted keys stay untouched.
	if shouldUpdate("secret_headers") {
		if secretHeaders, err = canonicalHeaders("secret_headers", secretHeaders); err != nil {
			return err
		}
		if err := validateHeaders("secret_headers", secretHeaders); err != nil {
			return err
		}
		mergedSecretHeaders, err := s.DecryptSecretHeaders(webhook.SecretHeaders)
		if err != nil {
			return fmt.Errorf("failed to decrypt existing secret headers: %w", err)
		}
		mergedSecretHeaders = canonicalNames(mergedSecretHeaders)
		if len(secretHeaders) == 0 {
			webhook.SecretHeaders = nil
		} else {
			if mergedSecretHeaders == nil {
				mergedSecretHeaders = map[string]string{}
			}
			for key, value := range secretHeaders {
				if value == "" {
					delete(mergedSecretHeaders, key)
					continue
				}
				mergedSecretHeaders[key] = value
			}
			encrypted, err := s.EncryptSecretHeaders(mergedSecretHeaders)
			if err != nil {
				return fmt.Errorf("failed to encrypt secret headers: %w", err)
			}
			webhook.SecretHeaders = encrypted
		}
	}
	if err := s.checkSecretHeadersFollowURL(oldURL, webhook.URL, oldSecretHeaders, secretHeaders, shouldUpdate("secret_headers")); err != nil {
		return err
	}
	// Update signature_type if in mask (or legacy non-empty)
	if shouldUpdate("signature_type") && signatureType != "" {
		newSigType := store.SignatureType(signatureType)
		if newSigType != store.SignatureTypeHMAC && newSigType != store.SignatureTypeEd25519 {
			return svcerrors.Errorf(svcerrors.InvalidArgument, "invalid signature_type: %q (must be \"hmac\" or \"ed25519\")", signatureType)
		}
		oldSigType := webhook.SignatureType
		webhook.SignatureType = newSigType
		// Generate Ed25519 keypair when switching to ed25519
		if newSigType == store.SignatureTypeEd25519 && oldSigType != store.SignatureTypeEd25519 {
			encPrivKey, err := s.generateEncryptedEd25519Key()
			if err != nil {
				return err
			}
			webhook.Ed25519PrivateKey = encPrivKey
		}
		// Clear Ed25519 key when switching away from ed25519
		if newSigType == store.SignatureTypeHMAC && oldSigType == store.SignatureTypeEd25519 {
			webhook.Ed25519PrivateKey = nil
		}
	}

	// Turning requires_transform on is checked inside the transaction, under
	// an exclusive lock that subscription writes (shared lock) wait on.
	turningOnTransform := mask["requires_transform"] && requiresTransform && !webhook.RequiresTransform
	if mask["requires_transform"] {
		webhook.RequiresTransform = requiresTransform
	}
	if replaceEvents && webhook.RequiresTransform {
		return svcerrors.Error(svcerrors.FailedPrecondition,
			"this webhook requires a payload transform, so its events cannot be replaced in bulk (that recreates subscriptions without templates); add or delete subscriptions one by one instead")
	}

	// Persist subscription replacement + webhook update atomically.
	err = s.webhookRepo.RunInTransaction(func(txRepo store.RepositoryInterface) error {
		if turningOnTransform {
			if _, err := txRepo.LockWebhook(ctx, tenantID, webhookUUID, consumer, true); err != nil {
				return fmt.Errorf("failed to lock webhook: %w", err)
			}
			subs, err := txRepo.ListSubscriptions(ctx, tenantID, webhookUUID)
			if err != nil {
				return fmt.Errorf("failed to list subscriptions: %w", err)
			}
			var missing []string
			for _, sub := range subs {
				if !sub.HasTransform() {
					missing = append(missing, sub.EventName)
				}
			}
			if len(missing) > 0 {
				return svcerrors.Errorf(svcerrors.FailedPrecondition,
					"cannot require a payload transform: the subscriptions for %s have none; add a transform_template to each first", strings.Join(missing, ", "))
			}
		}
		if replaceEvents {
			if err := txRepo.ReplaceWebhookSubscriptions(ctx, tenantID, webhookUUID, consumer, newSubs); err != nil {
				return fmt.Errorf("failed to update webhook subscriptions: %w", err)
			}
		}
		return txRepo.UpdateWebhook(ctx, tenantID, webhook)
	})
	if err != nil {
		s.logger.ErrorContext(ctx, "Failed to update webhook config",
			"webhook_id", webhookID,
			"error", err)
		return fmt.Errorf("failed to update webhook configuration: %w", err)
	}

	// Manage rate limit state: upsert when rate limit is set, delete when cleared
	if webhook.RateLimitRPS != nil {
		if err := s.webhookRepo.UpsertRateLimitState(ctx, tenantID, webhookUUID); err != nil {
			s.logger.WarnContext(ctx, "Failed to upsert rate limit state",
				"webhook_id", webhookID,
				"error", err,
			)
		}
	} else {
		// Rate limit removed (or was never set) — clean up any existing state
		if err := s.webhookRepo.DeleteRateLimitState(ctx, tenantID, webhookUUID); err != nil {
			s.logger.WarnContext(ctx, "Failed to delete rate limit state",
				"webhook_id", webhookID,
				"error", err,
			)
		}
	}

	s.logger.InfoContext(ctx, "Webhook configuration updated successfully",
		"webhook_id", webhookID)
	return nil
}

// RotateWebhookSecret replaces a webhook's HMAC signing secret with a freshly
// generated one and returns it in plaintext. This is the only time the new
// secret is readable; every later read masks it. Deliveries signed after the
// rotation use the new secret, so receivers must be updated with it.
func (s *WebhookService) RotateWebhookSecret(ctx context.Context, webhookID string, consumer string) (string, error) {
	if consumer == "" {
		return "", svcerrors.Error(svcerrors.InvalidArgument, "consumer is required")
	}
	webhookUUID, err := parseUUID(webhookID, "webhook ID")
	if err != nil {
		return "", err
	}

	webhook, err := s.webhookRepo.GetWebhookByID(ctx, tenant.FromContext(ctx), webhookUUID, consumer)
	if err != nil {
		return "", fmt.Errorf("failed to retrieve webhook: %w", err)
	}

	secret, err := generateWebhookSecret()
	if err != nil {
		return "", fmt.Errorf("failed to generate webhook secret: %w", err)
	}
	encSecret, err := s.EncryptWebhookSecret(secret)
	if err != nil {
		return "", fmt.Errorf("failed to encrypt webhook secret: %w", err)
	}
	webhook.WebhookSecret = encSecret

	if err := s.webhookRepo.UpdateWebhook(ctx, tenant.FromContext(ctx), webhook); err != nil {
		return "", fmt.Errorf("failed to store rotated webhook secret: %w", err)
	}
	s.logger.InfoContext(ctx, "Rotated webhook signing secret", "webhook_id", webhookID, "consumer", consumer)
	return secret, nil
}

// checkSecretHeadersFollowURL refuses to point a webhook at another origin
// while it keeps secret headers this request did not set. Secret header
// values are never returned by the API, so without this anyone who may edit
// the URL (a portal consumer, say) could read them by moving the webhook to
// a server they control. Each existing header must be given a new value, or
// "" to remove it; an empty secret_headers object removes them all.
func (s *WebhookService) checkSecretHeadersFollowURL(oldURL, newURL string, oldEncrypted []byte, set map[string]string, updated bool) error {
	if len(oldEncrypted) == 0 || sameOrigin(oldURL, newURL) {
		return nil
	}
	if updated && len(set) == 0 {
		return nil // all removed
	}
	old, err := s.DecryptSecretHeaders(oldEncrypted)
	if err != nil {
		return fmt.Errorf("failed to decrypt existing secret headers: %w", err)
	}
	var missing []string
	for name := range canonicalNames(old) {
		if _, ok := set[name]; !ok || !updated {
			missing = append(missing, name)
		}
	}
	if len(missing) == 0 {
		return nil
	}
	slices.Sort(missing)
	return svcerrors.Errorf(svcerrors.InvalidArgument,
		"changing the URL to another origin requires secret_headers to be set again: give %s a new value, or \"\" to remove it",
		strings.Join(missing, ", "))
}

// sameOrigin is client.SameOrigin for URL strings; one that does not parse
// is another origin.
func sameOrigin(a, b string) bool {
	ua, errA := neturl.Parse(a)
	ub, errB := neturl.Parse(b)
	if errA != nil || errB != nil {
		return false
	}
	return client.SameOrigin(ua, ub)
}