feedkit/pipeline/pipeline.go

package pipeline

import (
	"context"
	"fmt"

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

type Pipeline struct {
	Processors []processors.Processor
}

func (p *Pipeline) Process(ctx context.Context, e event.Event) (*event.Event, error) {
	if err := e.Validate(); err != nil {
		return nil, fmt.Errorf("pipeline: invalid input event: %w", err)
	}

	cur := &e

	for _, proc := range p.Processors {
		out, err := proc.Process(ctx, *cur)
		if err != nil {
			return nil, err
		}
		if out == nil {
			// Dropped by policy.
			return nil, nil
		}
		cur = out
	}

	if err := cur.Validate(); err != nil {
		return nil, fmt.Errorf("pipeline: invalid output event: %w", err)
	}

	return cur, nil
}