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 }