package webhooks import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "slices" "sort" "strings" "time" "github.com/google/uuid" "github.com/sarathsp06/sparrow" 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" "github.com/sarathsp06/sparrow/pkg/template" ) // Event type bundle: a JSON document holding many event type definitions, so // definitions can be exported from one environment and imported into // another. const ( BundleAPIVersion = "sparrow/v1" BundleKind = "EventTypeList" // BundleFormat is the bundle format this server writes and the newest it // reads. BundleFormat = 1 // MaxBundleItems caps how many event types one import may contain, which // bounds the import transaction. MaxBundleItems = 500 ) // Warning codes for an import's version stamp. Each is acknowledged by name. const ( StampVersionDiffers = "version_differs" StampFormatUnsupported = "format_unsupported" StampItemsChanged = "items_changed" // BlockedByBreaking is reported in ImportResult.BlockedBy when an item's // schema change is breaking and AllowBreaking was not set. BlockedByBreaking = "breaking" ) // BundleStamp records which Sparrow produced a bundle. It is a compatibility // hint, not a signature: anyone can edit a file and recompute it. type BundleStamp struct { SparrowVersion string Format int // SHA256 is the hex digest of the canonical encoding of the items. SHA256 string } // EventTypeExportSelection says which event types to export. Exactly one of // Names, Prefix or All is used. type EventTypeExportSelection struct { Names []string Prefix string All bool } // EventTypeImportOptions control an import. type EventTypeImportOptions struct { // DryRun computes the full result and writes nothing. DryRun bool // Acknowledge lists the stamp warning codes the caller accepts. Acknowledge []string // AllowBreaking applies breaking schema changes to subscribed types. AllowBreaking bool // PauseFailing pauses, in the same transaction, every subscription whose // template fails the pre-check against the new version. Off by default: // a mismatched template then fails its deliveries visibly instead. PauseFailing bool } // StampCheck is the outcome of comparing an import's stamp with this server. type StampCheck struct { // Status is "valid", "unsigned" (no stamp) or "warning". Status string ExportedBy string Server string Warnings []string } // TemplateCheckFailure is a subscription whose transform template no longer // renders against the new version of the event type. type TemplateCheckFailure struct { SubscriptionID string Consumer string WebhookID string // Payload is "sample" (every field) or "required_only". Payload string Error string } // TemplateCheck summarizes how an event type's subscriptions fare against a // new version. type TemplateCheck struct { Failures []TemplateCheckFailure Passed int WithoutTransform int CatchAll int // Paused lists subscriptions this import paused (PauseFailing). Paused []string } // EventTypeImportItem is the result for one bundle entry. type EventTypeImportItem struct { *EventTypeSaveResult // Templates is set for new versions. Templates *TemplateCheck } // EventTypeImportResult describes an import, applied or not. type EventTypeImportResult struct { Applied bool DryRun bool ImportedAt time.Time Stamp StampCheck Items []EventTypeImportItem // BlockedBy lists why nothing was written: unacknowledged stamp warning // codes, and BlockedByBreaking. BlockedBy []string } // bundleItem is the canonical encoding of one definition, used for the stamp // digest. Field order is fixed and Go sorts map keys, so equal definitions // always encode to the same bytes. type bundleItem struct { Name string `json:"name"` Description string `json:"description,omitempty"` Schema map[string]any `json:"event_schema,omitempty"` Metadata map[string]string `json:"metadata,omitempty"` Active bool `json:"active"` } // BundleDigest returns the stamp digest of a list of definitions: sha256 of // their canonical JSON encoding. Formatting of the source file does not // affect it. func BundleDigest(items []EventTypeDefinition) string { canon := make([]bundleItem, len(items)) for i, it := range items { canon[i] = bundleItem(it) if len(canon[i].Schema) == 0 { canon[i].Schema = nil } if len(canon[i].Metadata) == 0 { canon[i].Metadata = nil } } b, _ := json.Marshal(canon) // only maps, strings and bools: cannot fail sum := sha256.Sum256(b) return hex.EncodeToString(sum[:]) } // ExportEventTypes returns the current definitions of the selected event // types, sorted by name, and the stamp for the bundle. System event types // (sparrow.*) are never exported. An unknown name fails the whole export. func (s *WebhookService) ExportEventTypes(ctx context.Context, sel EventTypeExportSelection) ([]EventTypeDefinition, BundleStamp, error) { selectors := 0 if len(sel.Names) > 0 { selectors++ } if sel.Prefix != "" { selectors++ } if sel.All { selectors++ } if selectors != 1 { return nil, BundleStamp{}, svcerrors.Error(svcerrors.InvalidArgument, "choose exactly one of names, prefix or all") } tenantID := tenant.FromContext(ctx) var regs []*store.EventRegistration if len(sel.Names) > 0 { for _, name := range sel.Names { if IsReservedEventName(name) { return nil, BundleStamp{}, reservedEventNameError(name) } reg, err := s.webhookRepo.GetEventByName(ctx, tenantID, name) if err != nil { return nil, BundleStamp{}, fmt.Errorf("failed to load event type: %w", err) } if reg == nil { return nil, BundleStamp{}, svcerrors.Errorf(svcerrors.NotFound, "event type %q not found", name) } regs = append(regs, reg) } } else { for offset := 0; ; offset += maxPageLimit { page, _, err := s.webhookRepo.ListEventsPaginated(ctx, tenantID, store.EventTypeFilter{System: new(bool)}, maxPageLimit, offset) if err != nil { return nil, BundleStamp{}, fmt.Errorf("failed to list event types: %w", err) } for _, reg := range page { if !strings.HasPrefix(reg.Name, sel.Prefix) { continue } regs = append(regs, reg) } if len(page) < maxPageLimit { break } } } defs := make([]EventTypeDefinition, 0, len(regs)) seen := map[string]bool{} for _, reg := range regs { if seen[reg.Name] { continue } seen[reg.Name] = true defs = append(defs, EventTypeDefinition{ Name: reg.Name, Description: reg.Description, Schema: reg.Schema, Metadata: reg.Metadata, Active: reg.Active, }) } sort.Slice(defs, func(i, j int) bool { return defs[i].Name < defs[j].Name }) return defs, BundleStamp{SparrowVersion: sparrow.Version, Format: BundleFormat, SHA256: BundleDigest(defs)}, nil } // checkStamp compares an import's stamp with this server. func checkStamp(stamp *BundleStamp, items []EventTypeDefinition) StampCheck { check := StampCheck{Server: sparrow.Version} if stamp == nil { check.Status = "unsigned" return check } check.ExportedBy = stamp.SparrowVersion if stamp.SparrowVersion != sparrow.Version { check.Warnings = append(check.Warnings, StampVersionDiffers) } if stamp.Format < 1 || stamp.Format > BundleFormat { check.Warnings = append(check.Warnings, StampFormatUnsupported) } if stamp.SHA256 != BundleDigest(items) { check.Warnings = append(check.Warnings, StampItemsChanged) } check.Status = "valid" if len(check.Warnings) > 0 { check.Status = "warning" } return check } // validateBundle rejects a bundle that cannot be imported at all. Every // problem is reported, so a file can be fixed in one pass. func validateBundle(items []EventTypeDefinition) error { if len(items) == 0 { return svcerrors.Error(svcerrors.InvalidArgument, "bundle has no items") } if len(items) > MaxBundleItems { return svcerrors.Errorf(svcerrors.InvalidArgument, "bundle has %d items; at most %d are allowed per import", len(items), MaxBundleItems) } var problems []string seen := map[string]bool{} for _, it := range items { if err := validateEventTypeName(it.Name); err != nil { problems = append(problems, fmt.Sprintf("%q: %s", it.Name, svcerrors.Classify(err, "").ClientMessage())) continue } if seen[it.Name] { problems = append(problems, fmt.Sprintf("%q: appears more than once", it.Name)) } seen[it.Name] = true if len(it.Schema) > 0 { if err := compileSchema(it.Schema); err != nil { problems = append(problems, fmt.Sprintf("%q: event_schema does not compile: %v", it.Name, err)) } } } if len(problems) > 0 { return svcerrors.Errorf(svcerrors.InvalidArgument, "bundle is invalid, nothing was imported: %s", strings.Join(problems, "; ")) } return nil } func compileSchema(schema map[string]any) error { b, err := json.Marshal(schema) if err != nil { return err } _, err = newSchemaCompiler().Compile(b) return err } // errImportRollback aborts the import transaction after the result has been // computed (dry run, or blocked). var errImportRollback = errors.New("import rolled back") // ImportEventTypes applies a bundle in one transaction: every item is saved // with the usual rules (see planEventTypeSave), or nothing is. Types absent // from the bundle are never touched: an import does not delete. // // The result is computed in full either way. Nothing is written when DryRun // is set, when a stamp warning is not in Acknowledge, or when an item's // schema change is breaking for its subscriptions and AllowBreaking is not // set; BlockedBy then says why. func (s *WebhookService) ImportEventTypes(ctx context.Context, items []EventTypeDefinition, stamp *BundleStamp, opts EventTypeImportOptions) (*EventTypeImportResult, error) { if err := validateBundle(items); err != nil { return nil, err } sorted := slices.Clone(items) // Lock rows in name order so overlapping imports cannot deadlock. sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) tenantID := tenant.FromContext(ctx) result := &EventTypeImportResult{DryRun: opts.DryRun, Stamp: checkStamp(stamp, items)} for _, w := range result.Stamp.Warnings { if !slices.Contains(opts.Acknowledge, w) { result.BlockedBy = append(result.BlockedBy, w) } } // Render template checks before the transaction: a slow template can // take seconds per render, which must not hold row locks and a // connection. The transaction reuses a check when it sees the same // subscriptions, and renders again only if they changed in between. // Only types whose schema changes get a check, so an import of mostly // unchanged types renders nothing; a type that turns out to need a new // version after all (changed in between) is checked inside. prechecked := make(map[string]precheck, len(sorted)) for _, def := range sorted { current, err := s.webhookRepo.GetEventByName(ctx, tenantID, def.Name) if err != nil { return nil, fmt.Errorf("%s: failed to load event type: %w", def.Name, err) } if current == nil || planEventTypeSave(current, def).Action != EventTypeNewVersion { continue } subs, err := s.webhookRepo.ListSubscriptionsTargetingEvent(ctx, tenantID, def.Name) if err != nil { return nil, fmt.Errorf("%s: failed to list affected subscriptions: %w", def.Name, err) } prechecked[def.Name] = precheck{key: subscriptionsKey(subs), check: checkTemplates(def, subs)} } run := func() error { result.Items = result.Items[:0] result.BlockedBy = slices.DeleteFunc(result.BlockedBy, func(b string) bool { return b == BlockedByBreaking }) return s.webhookRepo.RunInTransaction(func(tx store.RepositoryInterface) error { breaking := false for _, def := range sorted { res, err := s.saveEventTypeTx(ctx, tx, tenantID, def, saveUpsert, saveEventTypeOptions{AllowBreaking: opts.AllowBreaking}) if err != nil { return fmt.Errorf("%s: %w", def.Name, err) } item := EventTypeImportItem{EventTypeSaveResult: res} if res.Action == EventTypeNewVersion { item.Templates = prechecked[def.Name].reuse(def, res.AffectedSubscriptions) if opts.PauseFailing && !res.Blocked { if err := pauseFailing(ctx, tx, tenantID, res, item.Templates); err != nil { return err } } } breaking = breaking || res.Blocked result.Items = append(result.Items, item) } if breaking { result.BlockedBy = append(result.BlockedBy, BlockedByBreaking) } if opts.DryRun || len(result.BlockedBy) > 0 { return errImportRollback } return nil }) } err := run() if errors.Is(err, storage.ErrAlreadyExists) { // A concurrent writer created one of the types first; the retry sees it. err = run() } switch { case errors.Is(err, errImportRollback): return result, nil case err != nil: return nil, err } result.Applied = true result.ImportedAt = time.Now().UTC() s.logger.InfoContext(ctx, "Imported event types", "count", len(result.Items)) return result, nil } // precheck is a template check computed before the import transaction, for // the subscriptions identified by key. type precheck struct { key string check *TemplateCheck } // reuse returns a copy of the precomputed check when subs are the ones it // was computed for, and checks subs afresh otherwise. func (p precheck) reuse(def EventTypeDefinition, subs []*store.EventSubscription) *TemplateCheck { if p.check == nil || p.key != subscriptionsKey(subs) { return checkTemplates(def, subs) } c := *p.check c.Failures = slices.Clone(p.check.Failures) c.Paused = nil return &c } // subscriptionsKey identifies what a template check depends on: each // subscription, its event and its transform. func subscriptionsKey(subs []*store.EventSubscription) string { h := sha256.New() for _, sub := range subs { _, _ = fmt.Fprintf(h, "%s\x00%s\x00%t\x00%d:%s\x00", sub.ID, sub.EventName, sub.TransformEnabled, len(sub.TransformTemplate), sub.TransformTemplate) } return hex.EncodeToString(h.Sum(nil)) } // pauseFailing pauses each subscription that failed the template check, // recording the import as the reason. func pauseFailing(ctx context.Context, tx store.RepositoryInterface, tenantID uuid.UUID, res *EventTypeSaveResult, check *TemplateCheck) error { reason := fmt.Sprintf("event type %s moved to v%d by import; its template failed the pre-check", res.Name, res.Version) seen := map[string]bool{} for _, f := range check.Failures { if seen[f.SubscriptionID] { continue } seen[f.SubscriptionID] = true id, err := uuid.Parse(f.SubscriptionID) if err != nil { return err } if _, err := tx.SetSubscriptionPaused(ctx, tenantID, id, true, reason); err != nil { return fmt.Errorf("pause subscription %s: %w", id, err) } check.Paused = append(check.Paused, f.SubscriptionID) } return nil } // checkTemplates renders every affected subscription's transform against the // new version, strictly, with a full sample payload and with only the // required fields. The second catches templates that read an optional field // without guarding for its absence. func checkTemplates(def EventTypeDefinition, subs []*store.EventSubscription) *TemplateCheck { check := &TemplateCheck{} if len(subs) == 0 { return check } payloads := []struct { name string data map[string]any }{ {"sample", generatePayload(def.Schema, true)}, {"required_only", generatePayload(def.Schema, false)}, } engine := template.New() strict := template.ExecOptions{StrictMissingKeys: true} for _, sub := range subs { if sub.EventName == store.CatchAllEventName { check.CatchAll++ } if !sub.TransformEnabled || sub.TransformTemplate == "" { check.WithoutTransform++ continue } failed := false for _, p := range payloads { ctx := template.NewWebhookContext("import-check", def.Name, time.Now().UTC().Format(time.RFC3339), 1, p.data) if _, err := engine.ExecuteWith(sub.TransformTemplate, ctx, strict); err != nil { check.Failures = append(check.Failures, TemplateCheckFailure{ SubscriptionID: sub.ID.String(), Consumer: sub.Consumer, WebhookID: sub.WebhookID.String(), Payload: p.name, Error: err.Error(), }) failed = true break } } if !failed { check.Passed++ } } return check } // generatePayload is store.SamplePayload, seeded so the import check is // repeatable. Generation errors yield an empty payload, which the strict // render then exercises. func generatePayload(schema map[string]any, all bool) map[string]any { sample, err := store.SamplePayload(schema, all, 1) if err != nil { return map[string]any{} } return sample }