// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

package xprocessorhelper // import "go.opentelemetry.io/collector/processor/processorhelper/xprocessorhelper"

import (
	"context"
	"errors"

	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/consumer/xconsumer"
	"go.opentelemetry.io/collector/pdata/pprofile"
	"go.opentelemetry.io/collector/processor"
	"go.opentelemetry.io/collector/processor/processorhelper"
	"go.opentelemetry.io/collector/processor/xprocessor"
)

// ProcessProfilesFunc is a helper function that processes the incoming data and returns the data to be sent to the next component.
// If error is returned then returned data are ignored. It MUST not call the next component.
type ProcessProfilesFunc func(context.Context, pprofile.Profiles) (pprofile.Profiles, error)

type profiles struct {
	component.StartFunc
	component.ShutdownFunc
	xconsumer.Profiles
}

// NewProfiles creates a xprocessor.Profiles that ensure context propagation.
func NewProfiles(
	_ context.Context,
	_ processor.Settings,
	_ component.Config,
	nextConsumer xconsumer.Profiles,
	profilesFunc ProcessProfilesFunc,
	options ...Option,
) (xprocessor.Profiles, error) {
	if profilesFunc == nil {
		return nil, errors.New("nil profilesFunc")
	}

	bs := fromOptions(options)
	profilesConsumer, err := xconsumer.NewProfiles(func(ctx context.Context, pd pprofile.Profiles) (err error) {
		pd, err = profilesFunc(ctx, pd)
		if err != nil {
			if errors.Is(err, processorhelper.ErrSkipProcessingData) {
				return nil
			}
			return err
		}
		return nextConsumer.ConsumeProfiles(ctx, pd)
	}, bs.consumerOptions...)
	if err != nil {
		return nil, err
	}

	return &profiles{
		StartFunc:    bs.StartFunc,
		ShutdownFunc: bs.ShutdownFunc,
		Profiles:     profilesConsumer,
	}, nil
}