feedkit/normalize/registry.go

package normalize

import (
	"context"
	"fmt"
	"sync"

	"gitea.maximumdirect.net/ejr/feedkit/event"
)

// Registry holds a set of Normalizers and selects one for a given event.
//
// Selection rule (simple + predictable):
//   - iterate in registration order
//   - the FIRST Normalizer whose Match(e) returns true is used
//
// If none match, the event passes through unchanged.
//
// Why "first match wins"?
// Normalization is usually a single mapping step from a raw schema/version into
// a normalized schema/version. If you want multiple transformation steps,
// model them as multiple pipeline processors (which feedkit already supports).
type Registry struct {
	mu sync.RWMutex
	ns []Normalizer
}

// Register adds a normalizer to the registry.
//
// Register panics if n is nil; this is a programmer error and should fail fast.
func (r *Registry) Register(n Normalizer) {
	if n == nil {
		panic("normalize.Registry.Register: normalizer cannot be nil")
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.ns = append(r.ns, n)
}

// Normalize finds the first matching Normalizer and applies it.
//
// If no normalizer matches, it returns the input event unchanged.
//
// If a normalizer returns (nil, nil), the event is dropped.
func (r *Registry) Normalize(ctx context.Context, in event.Event) (*event.Event, error) {
	if r == nil {
		// Nil registry is a valid "feature off" state.
		out := in
		return &out, nil
	}

	r.mu.RLock()
	ns := append([]Normalizer(nil), r.ns...) // copy for safe iteration outside lock
	r.mu.RUnlock()

	for _, n := range ns {
		if n == nil {
			// Shouldn't happen (Register panics), but guard anyway.
			continue
		}
		if !n.Match(in) {
			continue
		}

		out, err := n.Normalize(ctx, in)
		if err != nil {
			return nil, fmt.Errorf("normalize: normalizer failed: %w", err)
		}
		// out may be nil to signal "drop".
		return out, nil
	}

	// No match: pass through unchanged.
	out := in
	return &out, nil
}

// Processor adapts a Registry into a pipeline Processor.
//
// It implements:
//
//	Process(ctx context.Context, in event.Event) (*event.Event, error)
//
// which matches feedkit/pipeline.Processor.
//
// Optionality:
//   - If Registry is nil, Processor becomes a no-op pass-through.
//   - If Registry has no matching normalizer for an event, that event passes through unchanged.
type Processor struct {
	Registry *Registry

	// If true, events that do not match any normalizer cause an error.
	// Default is false (pass-through), which is the behavior you asked for.
	RequireMatch bool
}

// Process implements the pipeline.Processor interface.
func (p Processor) Process(ctx context.Context, in event.Event) (*event.Event, error) {
	// "Feature off": no registry means no normalization.
	if p.Registry == nil {
		out := in
		return &out, nil
	}

	out, err := p.Registry.Normalize(ctx, in)
	if err != nil {
		return nil, err
	}

	if out == nil {
		// Dropped by normalization policy.
		return nil, nil
	}

	if p.RequireMatch {
		// Detect "no-op pass-through due to no match" by checking whether a match existed.
		// We do this with a cheap second pass to avoid changing Normalize()'s signature.
		// (This is rare to enable; correctness/clarity > micro-optimization.)
		if !p.Registry.hasMatch(in) {
			return nil, fmt.Errorf("normalize: no normalizer matched event (id=%s kind=%s source=%s schema=%q)",
				in.ID, in.Kind, in.Source, in.Schema)
		}
	}

	return out, nil
}

func (r *Registry) hasMatch(in event.Event) bool {
	if r == nil {
		return false
	}
	r.mu.RLock()
	defer r.mu.RUnlock()
	for _, n := range r.ns {
		if n != nil && n.Match(in) {
			return true
		}
	}
	return false
}