internal/webhooks/event_type_save.go

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

package webhooks

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"reflect"
	"regexp"
	"strings"
	"unicode/utf8"

	"github.com/google/uuid"

	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/queue"
	"github.com/sarathsp06/sparrow/internal/webhooks/store"
)

// ReservedEventPrefix is the event name prefix reserved for Sparrow's own
// system events (e.g. sparrow.webhook.health_changed). Users can subscribe to
// these events but cannot register, change, import or push them.
const ReservedEventPrefix = store.SystemEventPrefix

// IsReservedEventName reports whether name uses the reserved prefix. The match
// is case-insensitive so "Sparrow.x" cannot slip past it.
func IsReservedEventName(name string) bool {
	return strings.HasPrefix(strings.ToLower(name), ReservedEventPrefix)
}

// EventTypeDefinition is the desired state of an event type: what a register,
// patch or import asks the event type to become.
type EventTypeDefinition struct {
	Name        string
	Description string
	Schema      map[string]any
	Metadata    map[string]string
	Active      bool
}

// EventTypeSaveAction is the outcome of saving an event type definition.
type EventTypeSaveAction string

const (
	// EventTypeCreated means the event type did not exist and was created at version 1.
	EventTypeCreated EventTypeSaveAction = "created"
	// EventTypeNewVersion means the schema changed and a new version was created.
	EventTypeNewVersion EventTypeSaveAction = "new_version"
	// EventTypeUpdated means the current version was changed in place: its
	// description, metadata or active flag, or a first schema filled in.
	EventTypeUpdated EventTypeSaveAction = "updated"
	// EventTypeUnchanged means the definition already matched; nothing was written.
	EventTypeUnchanged EventTypeSaveAction = "unchanged"
)

// Field names reported in EventTypeSaveResult.Changes.
const (
	ChangeSchema        = "schema"
	ChangeSchemaDefined = "schema_defined"
	ChangeDescription   = "description"
	ChangeMetadata      = "metadata"
	ChangeActive        = "active"
)

// EventTypeSaveResult describes what saving a definition did.
type EventTypeSaveResult struct {
	Name            string
	Action          EventTypeSaveAction
	Version         int
	PreviousVersion int
	Changes         []string
	// ActiveChange is "deactivates" or "reactivates" when the active flag
	// flips, and empty otherwise.
	ActiveChange string
	// Event is the resulting current definition. For a blocked save it is
	// the unchanged current definition.
	Event *store.EventRegistration
	// Compatibility is set for new_version: how the schema change looks from
	// a subscriber's point of view.
	Compatibility *SchemaCompatibility
	// AffectedSubscriptions are the subscriptions that receive this event
	// type (by name or catch-all), set for new_version.
	AffectedSubscriptions []*store.EventSubscription
	// Blocked is true when the change is breaking, subscriptions receive the
	// event type, and the caller did not allow breaking changes. Nothing was
	// written.
	Blocked bool
}

// saveEventTypeOptions are per-call choices for a save.
type saveEventTypeOptions struct {
	// AllowBreaking applies a breaking schema change even though
	// subscriptions receive the event type.
	AllowBreaking bool
}

// eventTypeSaveMode restricts which outcomes a save may have.
type eventTypeSaveMode int

const (
	// saveUpsert creates the event type or changes it (import).
	saveUpsert eventTypeSaveMode = iota
	// saveCreateOnly fails with AlreadyExists if the event type exists (register).
	saveCreateOnly
	// saveMustExist fails with NotFound if the event type does not exist (patch).
	saveMustExist
)

// planEventTypeSave decides what saving `in` over `current` should do,
// without touching storage. current is nil when the event type does not exist.
//
// Only a schema change creates a new version. Filling in a first schema on a
// version that had none defines the contract rather than changing it, so it
// updates that version in place. Description, metadata and active always
// change in place.
func planEventTypeSave(current *store.EventRegistration, in EventTypeDefinition) EventTypeSaveResult {
	res := EventTypeSaveResult{Name: in.Name}
	if current == nil {
		res.Action = EventTypeCreated
		res.Version = 1
		return res
	}

	res.PreviousVersion = current.Version
	res.Version = current.Version

	curEmpty, inEmpty := len(current.Schema) == 0, len(in.Schema) == 0
	switch {
	case curEmpty && !inEmpty:
		res.Changes = append(res.Changes, ChangeSchemaDefined)
	case !schemasEqual(current.Schema, in.Schema):
		res.Changes = append(res.Changes, ChangeSchema)
		res.Version = current.Version + 1
	}
	if current.Description != in.Description {
		res.Changes = append(res.Changes, ChangeDescription)
	}
	if !metadataEqual(current.Metadata, in.Metadata) {
		res.Changes = append(res.Changes, ChangeMetadata)
	}
	if current.Active != in.Active {
		res.Changes = append(res.Changes, ChangeActive)
		if in.Active {
			res.ActiveChange = "reactivates"
		} else {
			res.ActiveChange = "deactivates"
		}
	}

	switch {
	case res.Version != current.Version:
		res.Action = EventTypeNewVersion
	case len(res.Changes) > 0:
		res.Action = EventTypeUpdated
	default:
		res.Action = EventTypeUnchanged
	}
	return res
}

// schemasEqual compares two decoded JSON Schemas by value, so key order and
// whitespace never count as a change. No schema and an empty schema are equal.
func schemasEqual(a, b map[string]any) bool {
	if len(a) == 0 && len(b) == 0 {
		return true
	}
	return reflect.DeepEqual(a, b)
}

func metadataEqual(a, b map[string]string) bool {
	if len(a) == 0 && len(b) == 0 {
		return true
	}
	return maps.Equal(a, b)
}

// validateEventTypeName checks a user-supplied event type name.
func validateEventTypeName(name string) error {
	if name == "" {
		return svcerrors.Error(svcerrors.InvalidArgument, "event name is required")
	}
	if utf8.RuneCountInString(name) > maxEventNameLength {
		return svcerrors.Error(svcerrors.InvalidArgument, eventNameTooLong)
	}
	if IsReservedEventName(name) {
		return reservedEventNameError(name)
	}
	return nil
}

// eventNamePattern is the shape of a new event type name: dot-separated
// parts of lowercase letters, digits, '_' and '-', like order.created or
// stripe.payment_intent.succeeded. Event types can never be deleted, so the
// rule keeps a typo or a stray space from becoming permanent.
var eventNamePattern = regexp.MustCompile(`^[a-z0-9_-]+(\.[a-z0-9_-]+)*$`)

// validateNewEventTypeName checks the name of an event type about to be
// created. It is not applied to types that already exist, so names saved
// before the rule keep working for pushes and updates.
func validateNewEventTypeName(name string) error {
	if eventNamePattern.MatchString(name) {
		return nil
	}
	return svcerrors.Errorf(svcerrors.InvalidArgument,
		"event name %q is invalid: use lowercase letters, digits, '_' and '-' in dot-separated parts, like \"order.created\"", name)
}

func reservedEventNameError(name string) error {
	return svcerrors.Errorf(svcerrors.InvalidArgument,
		"event name %q uses the reserved prefix %q, which is for Sparrow's own events", name, ReservedEventPrefix)
}

// checkEventForConsumer keeps Sparrow's system events and tenant events
// apart. Sparrow emits sparrow.* events only under the _sparrow consumer, and
// _sparrow never receives tenant events, so a subscription crossing that line
// would never fire. A catch-all subscription fits any consumer.
func checkEventForConsumer(consumer, event string) error {
	isSystemConsumer := consumer == queue.SystemEventConsumer
	if event == store.CatchAllEventName || isSystemConsumer == IsReservedEventName(event) {
		return nil
	}
	if isSystemConsumer {
		return svcerrors.Errorf(svcerrors.InvalidArgument,
			"consumer %q only receives Sparrow's own %s* events, not %q", consumer, ReservedEventPrefix, event)
	}
	return svcerrors.Errorf(svcerrors.InvalidArgument,
		"%q is a Sparrow system event, delivered only under the %q consumer", event, queue.SystemEventConsumer)
}

// eventTypesForConsumer returns the filter for the event types consumer can
// subscribe to (see checkEventForConsumer). Sparrow's own events are listed
// only for the system consumer, so no consumer (all consumers) gets tenant
// event types.
func eventTypesForConsumer(consumer string) *bool {
	system := consumer == queue.SystemEventConsumer
	return &system
}

// saveEventType validates def and saves it in its own transaction.
func (s *WebhookService) saveEventType(ctx context.Context, def EventTypeDefinition, mode eventTypeSaveMode, opts saveEventTypeOptions) (*EventTypeSaveResult, error) {
	if err := validateEventTypeName(def.Name); err != nil {
		return nil, err
	}
	tenantID := tenant.FromContext(ctx)

	var res *EventTypeSaveResult
	run := func() error {
		return s.webhookRepo.RunInTransaction(func(tx store.RepositoryInterface) error {
			var err error
			res, err = s.saveEventTypeTx(ctx, tx, tenantID, def, mode, opts)
			return err
		})
	}
	err := run()
	// Two writers can both see "does not exist" (there is no row to lock) and
	// race to create it; the loser gets a unique violation. For an upsert the
	// row now exists, so run again and apply the change on top of it.
	if errors.Is(err, storage.ErrAlreadyExists) && mode == saveUpsert {
		err = run()
	}
	if err != nil {
		return nil, err
	}
	if res.Blocked {
		return res, breakingChangeError(res)
	}
	return res, nil
}

// breakingChangeError explains why a save was refused.
func breakingChangeError(res *EventTypeSaveResult) error {
	return svcerrors.Errorf(svcerrors.FailedPrecondition,
		"schema change to %q is breaking for %d subscription(s): %s. Pass allow_breaking=true to apply it anyway",
		res.Name, len(res.AffectedSubscriptions), strings.Join(res.Compatibility.Reasons, "; "))
}

// saveEventTypeTx applies def inside the transaction tx. It locks the current
// row, so concurrent writers to one event type are serialized and cannot
// claim the same next version.
func (s *WebhookService) saveEventTypeTx(ctx context.Context, tx store.RepositoryInterface, tenantID uuid.UUID, def EventTypeDefinition, mode eventTypeSaveMode, opts saveEventTypeOptions) (*EventTypeSaveResult, error) {
	current, err := tx.GetEventByNameForUpdate(ctx, tenantID, def.Name)
	if err != nil {
		return nil, fmt.Errorf("failed to load event type: %w", err)
	}
	if current != nil && mode == saveCreateOnly {
		return nil, svcerrors.Errorf(svcerrors.AlreadyExists, "event type %q already exists", def.Name)
	}
	if current == nil && mode == saveMustExist {
		return nil, svcerrors.Error(svcerrors.NotFound, "event type not found")
	}
	if current == nil {
		if err := validateNewEventTypeName(def.Name); err != nil {
			return nil, err
		}
	}

	plan := planEventTypeSave(current, def)
	res := &plan

	// A schema change is checked from the subscribers' side before anything
	// is written. A breaking change with nobody receiving the type needs no
	// opt-in: there is nothing to break.
	if plan.Action == EventTypeNewVersion {
		compat := ClassifySchemaChange(current.Schema, def.Schema)
		res.Compatibility = &compat
		subs, err := tx.ListSubscriptionsTargetingEvent(ctx, tenantID, def.Name)
		if err != nil {
			return nil, fmt.Errorf("failed to list affected subscriptions: %w", err)
		}
		res.AffectedSubscriptions = subs
		if compat.Breaking && len(subs) > 0 && !opts.AllowBreaking {
			res.Blocked = true
			res.Event = current
			return res, nil
		}
	}

	next := &store.EventRegistration{
		Name:        def.Name,
		Description: def.Description,
		Schema:      def.Schema,
		Metadata:    def.Metadata,
		Active:      def.Active,
		Version:     plan.Version,
	}

	switch plan.Action {
	case EventTypeUnchanged:
		res.Event = current
		return res, nil

	case EventTypeCreated:
		next.SamplePayload = s.samplePayloadFor(ctx, def.Schema)
		if err := tx.RegisterEvent(ctx, tenantID, next); err != nil {
			return nil, fmt.Errorf("failed to register event type: %w", err)
		}

	case EventTypeNewVersion:
		next.SamplePayload = s.samplePayloadFor(ctx, def.Schema)
		next.CreatedAt = current.CreatedAt
		if err := tx.UpdateEvent(ctx, tenantID, next); err != nil {
			return nil, fmt.Errorf("failed to update event type: %w", err)
		}
		if err := tx.AddEventTypeVersion(ctx, tenantID, &store.EventRegistrationVersion{
			Name:          def.Name,
			Version:       plan.Version,
			Description:   def.Description,
			Schema:        def.Schema,
			SamplePayload: next.SamplePayload,
		}); err != nil {
			return nil, fmt.Errorf("failed to record event type version: %w", err)
		}

	case EventTypeUpdated:
		next.CreatedAt = current.CreatedAt
		fillIn := len(current.Schema) == 0 && len(def.Schema) != 0
		if fillIn {
			next.SamplePayload = s.samplePayloadFor(ctx, def.Schema)
		} else {
			// Same contract: keep the stored schema (identical by value) and
			// its sample payload.
			next.Schema = current.Schema
			next.SamplePayload = current.SamplePayload
		}
		if err := tx.UpdateEvent(ctx, tenantID, next); err != nil {
			return nil, fmt.Errorf("failed to update event type: %w", err)
		}
		if fillIn {
			if err := tx.FillInEventTypeVersion(ctx, tenantID, &store.EventRegistrationVersion{
				Name:          def.Name,
				Version:       plan.Version,
				Description:   def.Description,
				Schema:        def.Schema,
				SamplePayload: next.SamplePayload,
			}); err != nil {
				return nil, fmt.Errorf("failed to fill in event type schema: %w", err)
			}
		}
	}

	saved, err := tx.GetEventByName(ctx, tenantID, def.Name)
	if err != nil {
		return nil, fmt.Errorf("failed to reload event type: %w", err)
	}
	res.Event = saved
	return res, nil
}

// samplePayloadFor generates the example payload stored with a schema. A
// generation failure is not fatal: the definition is saved with an empty
// sample.
func (s *WebhookService) samplePayloadFor(ctx context.Context, schema map[string]any) map[string]any {
	sample, err := store.SamplePayload(schema, true, 0)
	if err != nil {
		s.logger.WarnContext(ctx, "Failed to generate sample payload, using empty payload", "error", err)
		return map[string]any{}
	}
	return sample
}