Eric Rakestraw
a6c133319a
Introduce an optional normalization stage for feedkit pipelines via the new normalize package. This adds: - normalize.Normalizer interface with flexible Match() semantics - normalize.Registry for ordered normalizer selection (first match wins) - normalize.Processor adapter implementing pipeline.Processor - Pass-through behavior when no normalizer matches (normalization is optional) - Func helper for ergonomic normalizer definitions Update root doc.go to fully document the normalization model, its role in the pipeline, recommended conventions (Schema-based matching, raw vs normalized events), and concrete wiring examples. The documentation now serves as a complete external-facing API specification for downstream daemons such as weatherfeeder. This change preserves feedkit’s non-framework philosophy while enabling a clean separation between data collection and domain normalization.
77 lines
2.2 KiB
Go
77 lines
2.2 KiB
Go
package normalize
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"gitea.maximumdirect.net/ejr/feedkit/event"
|
|
)
|
|
|
|
// Normalizer converts one event shape into another.
|
|
//
|
|
// A Normalizer is typically domain-owned code (weatherfeeder/newsfeeder/...)
|
|
// that knows how to interpret a specific upstream payload and produce a
|
|
// normalized payload.
|
|
//
|
|
// Normalizers are selected via Match(). The matching strategy is intentionally
|
|
// flexible: implementations may match on Schema, Kind, Source, or any other
|
|
// Event fields.
|
|
type Normalizer interface {
|
|
// Match reports whether this normalizer applies to the given event.
|
|
//
|
|
// Common patterns:
|
|
// - match on e.Schema (recommended for versioning)
|
|
// - match on e.Source (useful if Schema is empty)
|
|
// - match on (e.Kind + e.Source), etc.
|
|
Match(e event.Event) bool
|
|
|
|
// Normalize transforms the incoming event into a new (or modified) event.
|
|
//
|
|
// Return values:
|
|
// - (out, nil) where out != nil: emit the normalized event
|
|
// - (nil, nil): drop the event (treat as policy drop)
|
|
// - (nil, err): fail the pipeline
|
|
//
|
|
// Note: If you simply want to pass the event through unchanged, return &in.
|
|
Normalize(ctx context.Context, in event.Event) (*event.Event, error)
|
|
}
|
|
|
|
// Func is an ergonomic adapter that lets you define a Normalizer with functions.
|
|
//
|
|
// Example:
|
|
//
|
|
// n := normalize.Func{
|
|
// MatchFn: func(e event.Event) bool { return e.Schema == "raw.openweather.current.v1" },
|
|
// NormalizeFn: func(ctx context.Context, in event.Event) (*event.Event, error) {
|
|
// // ... map in.Payload -> normalized payload ...
|
|
// },
|
|
// }
|
|
type Func struct {
|
|
MatchFn func(e event.Event) bool
|
|
NormalizeFn func(ctx context.Context, in event.Event) (*event.Event, error)
|
|
|
|
// Optional: helps produce nicer panic/error messages if something goes wrong.
|
|
Name string
|
|
}
|
|
|
|
func (f Func) Match(e event.Event) bool {
|
|
if f.MatchFn == nil {
|
|
return false
|
|
}
|
|
return f.MatchFn(e)
|
|
}
|
|
|
|
func (f Func) Normalize(ctx context.Context, in event.Event) (*event.Event, error) {
|
|
if f.NormalizeFn == nil {
|
|
return nil, fmt.Errorf("normalize.Func(%s): NormalizeFn is nil", f.safeName())
|
|
}
|
|
return f.NormalizeFn(ctx, in)
|
|
}
|
|
|
|
func (f Func) safeName() string {
|
|
if f.Name == "" {
|
|
return "<unnamed>"
|
|
}
|
|
return f.Name
|
|
}
|