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

package internal

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"

	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/component/componenttest"
	"go.opentelemetry.io/collector/config/configoptional"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
	"go.opentelemetry.io/collector/exporter/exportertest"
	"go.opentelemetry.io/collector/featuregate"
	"go.opentelemetry.io/collector/pipeline"
)

func TestNewQueueSenderFailedRequestDropped(t *testing.T) {
	qSet := queuebatch.AllSettings[request.Request]{
		Signal:    pipeline.SignalMetrics,
		ID:        component.NewID(exportertest.NopType),
		Telemetry: componenttest.NewNopTelemetrySettings(),
	}
	logger, observed := observer.New(zap.ErrorLevel)
	qSet.Telemetry.Logger = zap.New(logger)
	qCfg := NewDefaultQueueConfig()
	be, err := NewQueueSender(
		qSet, qCfg, "", sender.NewSender(func(context.Context, request.Request) error { return errors.New("some error") }),
	)
	require.NoError(t, err)

	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
	require.NoError(t, be.Send(context.Background(), &requesttest.FakeRequest{Items: 2}))
	require.NoError(t, be.Shutdown(context.Background()))
	assert.Len(t, observed.All(), 1)
	assert.Equal(t, "Exporting failed. Dropping data.", observed.All()[0].Message)
}

func TestNewDefaultQueueConfigBatchFeatureGate(t *testing.T) {
	// The batch config is present but not enabled by default.
	qCfg := NewDefaultQueueConfig()
	require.False(t, qCfg.Batch.HasValue())
	require.Equal(t, int64(8192), qCfg.Batch.GetOrInsertDefault().MinSize)

	require.NoError(t, featuregate.GlobalRegistry().Set(metadata.PkgExporterhelperQueueBatchEnabledFeatureGate.ID(), true))
	defer func() {
		require.NoError(t, featuregate.GlobalRegistry().Set(metadata.PkgExporterhelperQueueBatchEnabledFeatureGate.ID(), false))
	}()

	qCfg = NewDefaultQueueConfig()
	require.True(t, qCfg.Batch.HasValue())
	require.Equal(t, int64(8192), qCfg.Batch.Get().MinSize)
	require.Equal(t, 200*time.Millisecond, qCfg.Batch.Get().FlushTimeout)
	require.Equal(t, request.SizerTypeItems, qCfg.Batch.Get().Sizer)
}

func TestQueueConfig_Validate(t *testing.T) {
	qCfg := NewDefaultQueueConfig()
	require.NoError(t, qCfg.Validate())

	qCfg.NumConsumers = 0
	require.EqualError(t, qCfg.Validate(), "`num_consumers` must be positive")

	qCfg = NewDefaultQueueConfig()
	qCfg.QueueSize = 0
	require.EqualError(t, qCfg.Validate(), "`queue_size` must be positive")

	// Confirm Validate doesn't return error with invalid config when feature is disabled
	noCfg := configoptional.None[queuebatch.Config]()
	assert.NoError(t, noCfg.Validate())
}
