// Copyright (C) 2025 CardinalHQ, Inc
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, version 3.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

package filereader

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"testing"
	"time"

	"github.com/cardinalhq/oteltools/signalbuilder"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/pmetric"

	"github.com/cardinalhq/lakerunner/pipeline"
	"github.com/cardinalhq/lakerunner/pipeline/wkk"
)

// Helper function to create float64 pointer
func floatPtr(f float64) *float64 {
	return &f
}

// Helper function to create simple synthetic metrics data for basic tests
func createSimpleSyntheticMetrics() []byte {
	builder := signalbuilder.NewMetricsBuilder()
	resourceMetrics := &signalbuilder.ResourceMetrics{
		Resource: map[string]any{
			"service.name": "simple-test-service",
		},
		ScopeMetrics: []signalbuilder.ScopeMetrics{
			{
				Name: "simple-test-meter",
				Metrics: []signalbuilder.Metric{
					{
						Name: "test_gauge",
						Type: "gauge",
						Gauge: &signalbuilder.GaugeMetric{
							DataPoints: []signalbuilder.NumberDataPoint{
								{
									Value:     42.0,
									Timestamp: time.Now().UnixNano(),
								},
							},
						},
					},
				},
			},
		},
	}

	err := builder.Add(resourceMetrics)
	if err != nil {
		panic(err)
	}

	metrics := builder.Build()
	marshaler := &pmetric.ProtoMarshaler{}
	data, err := marshaler.MarshalMetrics(metrics)
	if err != nil {
		panic(err)
	}

	return data
}

func TestIngestProtoMetrics_New_InvalidData(t *testing.T) {
	// Test with invalid protobuf data
	invalidData := []byte("not a protobuf")
	reader := bytes.NewReader(invalidData)

	opts := ReaderOptions{
		BatchSize: 1000,
	}
	_, err := NewIngestProtoMetricsReader(reader, opts)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "failed to parse proto to OTEL metrics")
}

func TestIngestProtoMetrics_New_EmptyData(t *testing.T) {
	// Test with empty data
	emptyReader := bytes.NewReader([]byte{})

	opts := ReaderOptions{
		BatchSize: 1000,
	}
	reader, err := NewIngestProtoMetricsReader(emptyReader, opts)
	// Empty data may create a valid but empty metrics object
	if err != nil {
		assert.Contains(t, err.Error(), "failed to parse proto to OTEL metrics")
	} else {
		// If no error, should still be able to use the reader
		require.NotNil(t, reader)
		defer func() { _ = reader.Close() }()

		// Reading from empty metrics should return EOF immediately
		batch, readErr := reader.Next(context.TODO())
		assert.True(t, errors.Is(readErr, io.EOF))
		assert.Nil(t, batch)
	}
}

func TestIngestProtoMetrics_EmptySlice(t *testing.T) {
	syntheticData := createSimpleSyntheticMetrics()
	opts := ReaderOptions{
		BatchSize: 1000,
	}
	reader, err := NewIngestProtoMetricsReader(bytes.NewReader(syntheticData), opts)
	require.NoError(t, err)
	defer func() { _ = reader.Close() }()

	// Read with Next should return a batch
	batch, err := reader.Next(context.TODO())
	assert.NoError(t, err)
	assert.NotNil(t, batch)
	assert.Equal(t, 1, batch.Len())
}

func TestIngestProtoMetrics_Close(t *testing.T) {
	syntheticData := createSimpleSyntheticMetrics()
	opts := ReaderOptions{
		BatchSize: 1000,
	}
	reader, err := NewIngestProtoMetricsReader(bytes.NewReader(syntheticData), opts)
	require.NoError(t, err)

	// Should be able to read before closing
	batch, err := reader.Next(context.TODO())
	require.NoError(t, err)
	require.NotNil(t, batch)
	require.Equal(t, 1, batch.Len(), "Should read exactly 1 datapoint row before closing")

	// Close should work
	err = reader.Close()
	assert.NoError(t, err)

	// Reading after close should return error
	_, err = reader.Next(context.TODO())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "closed")

	// Close should be idempotent
	err = reader.Close()
	assert.NoError(t, err)
}

func TestIngestProtoMetrics_RowReusedAndCleared(t *testing.T) {
	builder := signalbuilder.NewMetricsBuilder()
	ts := time.Now()
	resourceMetrics := &signalbuilder.ResourceMetrics{
		Resource: map[string]any{"service.name": "reuse-test"},
		ScopeMetrics: []signalbuilder.ScopeMetrics{{
			Name: "reuse-meter",
			Metrics: []signalbuilder.Metric{{
				Name: "test_gauge",
				Type: "gauge",
				Gauge: &signalbuilder.GaugeMetric{
					DataPoints: []signalbuilder.NumberDataPoint{
						{Value: 1.0, Timestamp: ts.UnixNano()},
						{Value: 2.0, Timestamp: ts.Add(time.Minute).UnixNano()},
					},
				},
			}},
		}},
	}
	require.NoError(t, builder.Add(resourceMetrics))

	metrics := builder.Build()
	marshaler := &pmetric.ProtoMarshaler{}
	data, err := marshaler.MarshalMetrics(metrics)
	require.NoError(t, err)

	opts := ReaderOptions{
		BatchSize: 1,
	}
	reader, err := NewIngestProtoMetricsReader(bytes.NewReader(data), opts)
	require.NoError(t, err)
	defer func() { _ = reader.Close() }()

	batch1, err := reader.Next(context.TODO())
	require.NoError(t, err)
	require.NotNil(t, batch1)
	require.Equal(t, 1, batch1.Len())

	// With the new Batch API, we can't directly modify rows
	// Test that we can access the row data correctly
	row1 := batch1.Get(0)
	require.NotNil(t, row1)

	batch2, err := reader.Next(context.TODO())
	require.NoError(t, err)
	require.NotNil(t, batch2)
	require.Equal(t, 1, batch2.Len())

	// Verify second batch has valid data
	row2 := batch2.Get(0)
	require.NotNil(t, row2)
	_, exists := row2[wkk.NewRowKey("temp")]
	assert.False(t, exists, "new batch row should not contain temp data")

	// With the new batched interface, we can't reliably test address reuse
	// since memory management may work differently
}

func TestIngestProtoMetrics_ExponentialHistogram(t *testing.T) {
	metrics := pmetric.NewMetrics()
	rm := metrics.ResourceMetrics().AppendEmpty()
	sm := rm.ScopeMetrics().AppendEmpty()
	m := sm.Metrics().AppendEmpty()
	m.SetName("eh_metric")
	eh := m.SetEmptyExponentialHistogram()
	dp := eh.DataPoints().AppendEmpty()

	ts := pcommon.NewTimestampFromTime(time.Unix(1, 0).UTC())
	dp.SetTimestamp(ts)
	dp.SetStartTimestamp(ts)
	dp.SetCount(6)
	dp.SetScale(1)
	dp.SetZeroCount(1)
	dp.SetSum(10)
	dp.SetMin(1)
	dp.SetMax(9)
	dp.SetFlags(pmetric.DefaultDataPointFlags.WithNoRecordedValue(true))

	pos := dp.Positive()
	pos.SetOffset(0)
	pos.BucketCounts().FromRaw([]uint64{1, 2})

	neg := dp.Negative()
	neg.SetOffset(-1)
	neg.BucketCounts().FromRaw([]uint64{3})

	ex := dp.Exemplars().AppendEmpty()
	ex.SetTimestamp(ts)
	ex.SetDoubleValue(5.5)
	ex.SetTraceID(pcommon.TraceID([16]byte{1}))
	ex.SetSpanID(pcommon.SpanID([8]byte{2}))
	ex.FilteredAttributes().PutStr("foo", "bar")

	opts := ReaderOptions{
		BatchSize: 1000,
	}
	reader, err := NewIngestProtoMetricsReaderFromMetrics(&metrics, opts)
	require.NoError(t, err)
	defer func() { _ = reader.Close() }()

	batch, err := reader.Next(context.TODO())

	// Exponential histogram processing is now implemented (though conversion is stubbed)
	// Zero count creates valid data, so expect successful batch
	assert.NoError(t, err)
	assert.NotNil(t, batch)
	assert.Equal(t, 1, batch.Len())

	// This test verifies that exponential histograms are processed through the conversion pipeline
	// Even with stubbed conversion, zero count creates valid sketch data
}

// Test parsing function directly with invalid data
func TestParseProtoToOtelMetrics_InvalidData(t *testing.T) {
	invalidData := bytes.NewReader([]byte("invalid protobuf data"))
	_, err := parseProtoToOtelMetrics(invalidData)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "failed to unmarshal protobuf metrics")
}

// Test parsing function with read error
func TestParseProtoToOtelMetrics_ReadError(t *testing.T) {
	errorReader := &errorReaderImpl{shouldError: true}
	_, err := parseProtoToOtelMetrics(errorReader)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "failed to read data")
}

// errorReaderImpl is a test reader that can simulate read errors
type errorReaderImpl struct {
	shouldError bool
}

func (e *errorReaderImpl) Read(p []byte) (int, error) {
	if e.shouldError {
		return 0, errors.New("simulated read error")
	}
	return 0, io.EOF
}

// Test ingest proto metrics with comprehensive synthetic metric data
func TestIngestProtoMetrics_SyntheticMultiTypeMetrics(t *testing.T) {
	builder := signalbuilder.NewMetricsBuilder()

	// Create comprehensive metrics with all major types
	resourceMetrics := &signalbuilder.ResourceMetrics{
		Resource: map[string]any{
			"service.name":    "synthetic-metrics-service",
			"service.version": "1.0.0",
			"host.name":       "test-host-01",
			"environment":     "testing",
		},
		ScopeMetrics: []signalbuilder.ScopeMetrics{
			{
				Name:    "synthetic-meter",
				Version: "1.0.0",
				Attributes: map[string]any{
					"meter.library": "synthetic",
				},
				Metrics: []signalbuilder.Metric{
					// Gauge metric
					{
						Name:        "cpu_usage_percent",
						Description: "Current CPU usage percentage",
						Unit:        "percent",
						Type:        "gauge",
						Gauge: &signalbuilder.GaugeMetric{
							DataPoints: []signalbuilder.NumberDataPoint{
								{
									Attributes: map[string]any{
										"cpu":  "cpu0",
										"host": "server-01",
									},
									Value:     85.5,
									Timestamp: time.Now().UnixNano(),
								},
								{
									Attributes: map[string]any{
										"cpu":  "cpu1",
										"host": "server-01",
									},
									Value:     72.3,
									Timestamp: time.Now().UnixNano(),
								},
							},
						},
					},
					// Sum metric (counter)
					{
						Name:        "http_requests_total",
						Description: "Total HTTP requests",
						Unit:        "1",
						Type:        "sum",
						Sum: &signalbuilder.SumMetric{
							IsMonotonic:            true,
							AggregationTemporality: "cumulative",
							DataPoints: []signalbuilder.NumberDataPoint{
								{
									Attributes: map[string]any{
										"method":   "GET",
										"status":   "200",
										"endpoint": "/api/users",
									},
									Value:     1250,
									Timestamp: time.Now().UnixNano(),
								},
								{
									Attributes: map[string]any{
										"method":   "POST",
										"status":   "201",
										"endpoint": "/api/users",
									},
									Value:     875,
									Timestamp: time.Now().UnixNano(),
								},
								{
									Attributes: map[string]any{
										"method":   "GET",
										"status":   "404",
										"endpoint": "/api/notfound",
									},
									Value:     45,
									Timestamp: time.Now().UnixNano(),
								},
							},
						},
					},
					// Histogram metric
					{
						Name:        "http_request_duration_seconds",
						Description: "HTTP request duration in seconds",
						Unit:        "s",
						Type:        "histogram",
						Histogram: &signalbuilder.HistogramMetric{
							AggregationTemporality: "cumulative",
							DataPoints: []signalbuilder.HistogramDataPoint{
								{
									Attributes: map[string]any{
										"method":   "GET",
										"endpoint": "/api/users",
									},
									Count:          100,
									Sum:            floatPtr(45.67),
									BucketCounts:   []uint64{10, 25, 35, 20, 8, 2},
									ExplicitBounds: []float64{0.1, 0.5, 1.0, 2.0, 5.0},
									Timestamp:      time.Now().UnixNano(),
								},
								{
									Attributes: map[string]any{
										"method":   "POST",
										"endpoint": "/api/orders",
									},
									Count:          75,
									Sum:            floatPtr(89.23),
									BucketCounts:   []uint64{5, 15, 25, 20, 10},
									ExplicitBounds: []float64{0.1, 0.5, 1.0, 2.0},
									Timestamp:      time.Now().UnixNano(),
								},
							},
						},
					},
				},
			},
		},
	}

	err := builder.Add(resourceMetrics)
	require.NoError(t, err, "Should successfully add synthetic metrics")

	// Build and test
	metrics := builder.Build()
	marshaler := &pmetric.ProtoMarshaler{}
	data, err := marshaler.MarshalMetrics(metrics)
	require.NoError(t, err, "Should successfully marshal metrics")

	reader := bytes.NewReader(data)
	opts := ReaderOptions{
		BatchSize: 1000,
	}
	protoReader, err := NewIngestProtoMetricsReader(reader, opts)
	require.NoError(t, err)
	require.NotNil(t, protoReader)
	defer func() { _ = protoReader.Close() }()

	// Read all rows
	allRows, err := readAllRows(protoReader)
	require.NoError(t, err)
	require.Equal(t, 7, len(allRows), "Should read exactly 7 datapoint rows (2 gauge + 3 sum + 2 histogram)")

	// Verify different metric types are present
	metricTypes := make(map[string]int)
	metricNames := make(map[string]int)

	for _, row := range allRows {
		if metricType, exists := row[wkk.RowKeyCMetricType]; exists {
			if typeStr, ok := metricType.(string); ok {
				metricTypes[typeStr]++
			}
		}
		if metricName, exists := row[wkk.RowKeyCName]; exists {
			if nameStr, ok := metricName.(string); ok {
				metricNames[nameStr]++
			}
		}

		// All rows should have resource attributes
		assert.Equal(t, "synthetic-metrics-service", row[wkk.NewRowKey("resource_service_name")])
		assert.Equal(t, "1.0.0", row[wkk.NewRowKey("resource_service_version")])
		assert.Equal(t, "test-host-01", row[wkk.NewRowKey("resource_host_name")])
		assert.Equal(t, "testing", row[wkk.NewRowKey("resource_environment")])

		// All rows should have scope attributes
		assert.Equal(t, "synthetic", row[wkk.NewRowKey("scope_meter_library")])
	}

	// Verify metric type distribution
	assert.Equal(t, 2, metricTypes["gauge"], "Should have 2 Gauge datapoints")
	assert.Equal(t, 3, metricTypes["count"], "Should have 3 Sum datapoints")
	assert.Equal(t, 2, metricTypes["histogram"], "Should have 2 Histogram datapoints")

	// Verify metric name distribution
	assert.Equal(t, 2, metricNames["cpu_usage_percent"], "Should have 2 CPU usage datapoints")
	assert.Equal(t, 3, metricNames["http_requests_total"], "Should have 3 HTTP request datapoints")
	assert.Equal(t, 2, metricNames["http_request_duration_seconds"], "Should have 2 duration datapoints")

	// Test batched reading with a new reader instance
	protoReader2, err := NewIngestProtoMetricsReader(bytes.NewReader(data), opts)
	require.NoError(t, err)
	defer func() { _ = protoReader2.Close() }()

	// Read in batches
	var totalBatchedRows int
	for {
		batch, readErr := protoReader2.Next(context.TODO())
		if batch != nil {
			totalBatchedRows += batch.Len()

			// Verify each row that was read
			for i := 0; i < batch.Len(); i++ {
				row := batch.Get(i)
				assert.Greater(t, len(row), 0, "Batched row %d should have data", i)
				_, hasName := row[wkk.RowKeyCName]
				assert.True(t, hasName)
				_, hasMetricType := row[wkk.RowKeyCMetricType]
				assert.True(t, hasMetricType)
			}
		}

		if errors.Is(readErr, io.EOF) {
			break
		}
		require.NoError(t, readErr)
	}
	assert.Equal(t, len(allRows), totalBatchedRows, "Batched reading should read same number of rows")

	// Test single row reading
	opts1 := ReaderOptions{
		BatchSize: 1,
	}
	protoReader3, err := NewIngestProtoMetricsReader(bytes.NewReader(data), opts1)
	require.NoError(t, err)
	defer func() { _ = protoReader3.Close() }()

	batch, err := protoReader3.Next(context.TODO())
	require.NoError(t, err)
	require.NotNil(t, batch)
	require.Equal(t, 1, batch.Len(), "Should read exactly 1 row in first batch")
	_, hasName := batch.Get(0)[wkk.RowKeyCName]
	assert.True(t, hasName)
	_, hasResourceServiceName := batch.Get(0)[wkk.NewRowKey("resource_service_name")]
	assert.True(t, hasResourceServiceName)

	// Test data exhaustion - continue reading until EOF
	var exhaustRows int
	for {
		batch, readErr := protoReader3.Next(context.TODO())
		if batch != nil {
			exhaustRows += batch.Len()
		}
		if errors.Is(readErr, io.EOF) {
			break
		}
		require.NoError(t, readErr)
	}
	// Should have read all remaining rows
	assert.Equal(t, len(allRows)-1, exhaustRows, "Should read all remaining rows after first single read")

	t.Logf("Successfully processed %d synthetic metric datapoints", len(allRows))
}

// Test ingest proto metrics with multi-resource synthetic data
func TestIngestProtoMetrics_SyntheticMultiResourceMetrics(t *testing.T) {
	builder := signalbuilder.NewMetricsBuilder()

	// Add metrics from multiple services/resources
	services := []struct {
		name     string
		resource map[string]any
		metrics  []signalbuilder.Metric
	}{
		{
			name: "web-service",
			resource: map[string]any{
				"service.name":           "web-frontend",
				"service.version":        "2.1.0",
				"deployment.environment": "production",
			},
			metrics: []signalbuilder.Metric{
				{
					Name: "active_connections",
					Type: "gauge",
					Unit: "1",
					Gauge: &signalbuilder.GaugeMetric{
						DataPoints: []signalbuilder.NumberDataPoint{
							{
								Attributes: map[string]any{"port": "8080"},
								Value:      125,
								Timestamp:  time.Now().UnixNano(),
							},
						},
					},
				},
				{
					Name: "requests_per_second",
					Type: "sum",
					Unit: "1/s",
					Sum: &signalbuilder.SumMetric{
						IsMonotonic: false,
						DataPoints: []signalbuilder.NumberDataPoint{
							{
								Attributes: map[string]any{"handler": "api"},
								Value:      45.6,
								Timestamp:  time.Now().UnixNano(),
							},
						},
					},
				},
			},
		},
		{
			name: "database-service",
			resource: map[string]any{
				"service.name":           "postgres-db",
				"service.version":        "13.7",
				"deployment.environment": "production",
			},
			metrics: []signalbuilder.Metric{
				{
					Name: "query_duration_ms",
					Type: "histogram",
					Unit: "ms",
					Histogram: &signalbuilder.HistogramMetric{
						AggregationTemporality: "cumulative",
						DataPoints: []signalbuilder.HistogramDataPoint{
							{
								Attributes:     map[string]any{"table": "users"},
								Count:          1000,
								Sum:            floatPtr(15000.5),
								BucketCounts:   []uint64{100, 400, 350, 125, 25},
								ExplicitBounds: []float64{1, 10, 50, 100},
								Timestamp:      time.Now().UnixNano(),
							},
						},
					},
				},
			},
		},
	}

	// Add each service's metrics
	for _, service := range services {
		resourceMetrics := &signalbuilder.ResourceMetrics{
			Resource: service.resource,
			ScopeMetrics: []signalbuilder.ScopeMetrics{
				{
					Name:    service.name + "-meter",
					Version: "1.0.0",
					Metrics: service.metrics,
				},
			},
		}
		err := builder.Add(resourceMetrics)
		require.NoError(t, err, "Should add metrics for %s", service.name)
	}

	// Build and test
	metrics := builder.Build()
	marshaler := &pmetric.ProtoMarshaler{}
	data, err := marshaler.MarshalMetrics(metrics)
	require.NoError(t, err)

	reader := bytes.NewReader(data)
	opts := ReaderOptions{
		BatchSize: 1000,
	}
	protoReader, err := NewIngestProtoMetricsReader(reader, opts)
	require.NoError(t, err)
	defer func() { _ = protoReader.Close() }()

	// Should read metrics from both services (3 total datapoints)
	allRows, err := readAllRows(protoReader)
	require.NoError(t, err)
	require.Equal(t, 3, len(allRows), "Should read datapoints from both services")

	// Count datapoints by service
	webServiceCount := 0
	dbServiceCount := 0
	for _, row := range allRows {
		serviceName := row[wkk.NewRowKey("resource_service_name")].(string)
		switch serviceName {
		case "web-frontend":
			webServiceCount++
		case "postgres-db":
			dbServiceCount++
		}
	}

	assert.Equal(t, 2, webServiceCount, "Should have 2 datapoints from web service")
	assert.Equal(t, 1, dbServiceCount, "Should have 1 datapoint from database service")

	t.Logf("Successfully processed %d datapoints from %d services", len(allRows), len(services))
}

// Test ingest proto metrics with edge case synthetic data
func TestIngestProtoMetrics_SyntheticEdgeCases(t *testing.T) {
	builder := signalbuilder.NewMetricsBuilder()

	resourceMetrics := &signalbuilder.ResourceMetrics{
		Resource: map[string]any{
			"service.name": "edge-case-service",
		},
		ScopeMetrics: []signalbuilder.ScopeMetrics{
			{
				Name: "edge-case-meter",
				Metrics: []signalbuilder.Metric{
					// Metric with no attributes
					{
						Name: "simple_counter",
						Type: "sum",
						Sum: &signalbuilder.SumMetric{
							IsMonotonic: true,
							DataPoints: []signalbuilder.NumberDataPoint{
								{
									Value:     100,
									Timestamp: time.Now().UnixNano(),
									// No attributes - testing empty attributes
								},
							},
						},
					},
					// Metric with zero value
					{
						Name: "zero_gauge",
						Type: "gauge",
						Gauge: &signalbuilder.GaugeMetric{
							DataPoints: []signalbuilder.NumberDataPoint{
								{
									Attributes: map[string]any{
										"measurement": "idle",
									},
									Value:     0.0,
									Timestamp: time.Now().UnixNano(),
								},
							},
						},
					},
					// Histogram with single bucket
					{
						Name: "simple_histogram",
						Type: "histogram",
						Histogram: &signalbuilder.HistogramMetric{
							AggregationTemporality: "cumulative",
							DataPoints: []signalbuilder.HistogramDataPoint{
								{
									Attributes: map[string]any{
										"operation": "test",
									},
									Count:          5,
									Sum:            floatPtr(12.5),
									BucketCounts:   []uint64{5},
									ExplicitBounds: []float64{}, // Single bucket, no bounds
									Timestamp:      time.Now().UnixNano(),
								},
							},
						},
					},
				},
			},
		},
	}

	err := builder.Add(resourceMetrics)
	require.NoError(t, err, "Should successfully add edge case metrics")

	// Build and test
	metrics := builder.Build()
	marshaler := &pmetric.ProtoMarshaler{}
	data, err := marshaler.MarshalMetrics(metrics)
	require.NoError(t, err)

	reader := bytes.NewReader(data)
	opts := ReaderOptions{
		BatchSize: 1000,
	}
	protoReader, err := NewIngestProtoMetricsReader(reader, opts)
	require.NoError(t, err)
	defer func() { _ = protoReader.Close() }()

	// Should read all edge case datapoints
	allRows, err := readAllRows(protoReader)
	require.NoError(t, err)
	require.Equal(t, 3, len(allRows), "Should read 3 edge case datapoints")

	// Verify edge cases
	for i, row := range allRows {
		t.Run(fmt.Sprintf("edge_case_%d", i), func(t *testing.T) {
			// All should have resource attributes
			assert.Equal(t, "edge-case-service", row[wkk.NewRowKey("resource_service_name")])

			metricName := row[wkk.RowKeyCName].(string)
			switch metricName {
			case "simple_counter":
				// Should handle metrics with no attributes gracefully
				assert.Equal(t, "count", row[wkk.RowKeyCMetricType])
				assert.Equal(t, 100.0, row[wkk.RowKeyRollupSum])
			case "zero_gauge":
				// Should handle zero values
				assert.Equal(t, "gauge", row[wkk.RowKeyCMetricType])
				assert.Equal(t, 0.0, row[wkk.RowKeyRollupSum])
				assert.Equal(t, "idle", row[wkk.NewRowKey("attr_measurement")])
			case "simple_histogram":
				// Should handle simple histograms
				assert.Equal(t, "histogram", row[wkk.RowKeyCMetricType])
				assert.Equal(t, "test", row[wkk.NewRowKey("attr_operation")])
			}
		})
	}

	t.Logf("Successfully processed %d edge case datapoints", len(allRows))
}

func TestIngestProtoMetrics_HistogramAlwaysHasSketch(t *testing.T) {
	// Test that the fix for empty histogram bug is working - histograms should always have sketches
	// This test verifies that histograms don't cause "Empty sketch without valid rollup_sum" errors

	syntheticData := createSimpleSyntheticMetrics()
	opts := ReaderOptions{
		BatchSize: 1000,
	}
	reader, err := NewIngestProtoMetricsReader(bytes.NewReader(syntheticData), opts)
	require.NoError(t, err)
	defer func() { _ = reader.Close() }()

	// This test mainly verifies that our renaming and basic functionality works
	// The real fix is tested in production where histograms with empty buckets
	// now always create sketches with rollup fields instead of causing errors

	// Read at least one row to verify reader works
	batch, err := reader.Next(context.TODO())
	if err != nil && err != io.EOF {
		t.Fatalf("Unexpected error reading: %v", err)
	}

	// Should have read the gauge metric from createSimpleSyntheticMetrics
	if batch != nil && batch.Len() > 0 {
		row := batch.Get(0)
		assert.Equal(t, "test_gauge", row[wkk.RowKeyCName])
		assert.Equal(t, "gauge", row[wkk.RowKeyCMetricType])
		t.Logf("Successfully read metric: %s", row[wkk.RowKeyCName])
	}
}

func TestIngestProtoMetrics_ContractCompliance(t *testing.T) {
	// Test that IngestProtoMetricsReader complies with the 11-point contract
	syntheticData := createSimpleSyntheticMetrics()
	opts := ReaderOptions{
		BatchSize: 1000,
	}
	reader, err := NewIngestProtoMetricsReader(bytes.NewReader(syntheticData), opts)
	require.NoError(t, err)
	defer func() { _ = reader.Close() }()

	// Read all available rows
	var allRows []pipeline.Row
	for {
		batch, err := reader.Next(context.TODO())
		if batch != nil {
			for i := 0; i < batch.Len(); i++ {
				allRows = append(allRows, batch.Get(i))
			}
		}

		if err == io.EOF {
			break
		}
		require.NoError(t, err)
	}

	// Validate contract compliance using the shared validation function
	ValidateMetricsReaderContract(t, allRows)

	t.Logf("Successfully validated %d rows against metrics reader contract", len(allRows))
}

// ValidateMetricsReaderContract validates that a reader produces metrics data conforming to the 12-point contract.
// This function can be used to test any metrics reader to ensure consistent output format.
// Contract requirements:
//  1. All rollup fields must always be set (chq_rollup_avg, chq_rollup_max, chq_rollup_min, chq_rollup_count, chq_rollup_sum, chq_rollup_p25, chq_rollup_p50, chq_rollup_p75, chq_rollup_p90, chq_rollup_p95, chq_rollup_p99)
//  2. All metric attribute tags must be prefixed with "attr_"
//  3. All resource attribute tags must be prefixed with "resource_"
//  4. All scope attribute tags must be prefixed with "scope_"
//  5. All metrics must have metric_name as a string
//  6. All metrics must have chq_metric_type as one of "gauge", "count", "histogram"
//  7. All metrics must have chq_timestamp as int64 (use Timestamp, fallback to StartTimestamp)
//  8. NO chq_value field should be present (removed from all metric types)
//  9. Summary data points should be silently dropped (not present in output)
//
// 10. All metrics must have scope_url, scope_name, description, unit, type fields
// 11. chq_tid computation should be done after transform (not in reader)
// 12. For histograms, len(chq_sketch) must be > 0; for gauge/count, len(chq_sketch) must be 0
func ValidateMetricsReaderContract(t *testing.T, rows []pipeline.Row) {
	require.NotEmpty(t, rows, "ValidateMetricsReaderContract requires at least one row to validate")

	for i, row := range rows {
		// 1. All rollup fields must always be set
		rollupFields := []string{"chq_rollup_avg", "chq_rollup_max", "chq_rollup_min", "chq_rollup_count", "chq_rollup_sum", "chq_rollup_p25", "chq_rollup_p50", "chq_rollup_p75", "chq_rollup_p90", "chq_rollup_p95", "chq_rollup_p99"}
		for _, field := range rollupFields {
			fieldKey := wkk.NewRowKey(field)
			val, hasRollupField := row[fieldKey]
			assert.True(t, hasRollupField, "Row %d missing rollup field %s", i, field)
			if hasRollupField {
				assert.IsType(t, float64(0), val, "Row %d rollup field %s must be float64", i, field)
			}
		}

		// 2-4. Validate attribute prefixing
		// Check that all attr/resource/scope attributes are properly prefixed
		expectedPrefixes := []string{"attr_", "resource_", "scope_"}
		standardFields := []string{"chq_description", "chq_unit", "chq_scope_url", "chq_scope_name", "chq_sketch"}

		for key := range row {
			// Skip CardinalHQ internal fields and standard literal fields
			keyStr := string(key.Value())
			if strings.HasPrefix(keyStr, "chq_") {
				continue
			}

			// Skip known literal fields that should not be prefixed
			if contains(standardFields, keyStr) {
				continue
			}

			// For any other field, it should have one of the expected prefixes
			hasValidPrefix := false
			for _, prefix := range expectedPrefixes {
				if strings.HasPrefix(keyStr, prefix) {
					hasValidPrefix = true
					break
				}
			}

			if !hasValidPrefix {
				t.Logf("Row %d: Field '%s' should have prefix (attr_, resource_, or scope_)", i, string(key.Value()))
			}
		}

		// 5. All metrics must have metric_name as string
		name, nameOk := row[wkk.RowKeyCName].(string)
		assert.True(t, nameOk, "Row %d missing or invalid metric_name field", i)
		assert.NotEmpty(t, name, "Row %d metric_name must not be empty", i)

		// 6. All metrics must have chq_metric_type as valid type
		metricType, metricTypeOk := row[wkk.RowKeyCMetricType].(string)
		assert.True(t, metricTypeOk, "Row %d missing or invalid chq_metric_type field", i)
		assert.Contains(t, []string{"gauge", "count", "histogram"}, metricType, "Row %d invalid chq_metric_type: %s", i, metricType)

		// 7. All metrics must have chq_timestamp as int64
		timestamp, timestampOk := row[wkk.RowKeyCTimestamp].(int64)
		assert.True(t, timestampOk, "Row %d missing or invalid chq_timestamp field", i)
		assert.Greater(t, timestamp, int64(0), "Row %d chq_timestamp must be positive", i)

		// 8. NO chq_value field should be present
		_, valueExists := row[wkk.RowKeyCValue]
		assert.False(t, valueExists, "Row %d should not contain chq_value field", i)

		// 9. Summary data points should be silently dropped (verified by absence in output)
		// This is implicitly validated since summaries return errors and are not included

		// 10. All metrics must have required fields
		requiredFields := []string{"chq_scope_url", "chq_scope_name", "chq_description", "chq_unit"}
		for _, field := range requiredFields {
			_, hasRequiredField := row[wkk.NewRowKey(field)]
			assert.True(t, hasRequiredField, "Row %d missing required field %s", i, field)
		}

		// 11. chq_tid computation should be done after transform (not in reader)
		// The reader should NOT include TID - it should be added by MetricTranslator
		_, tidExists := row[wkk.RowKeyCTID]
		assert.False(t, tidExists, "Row %d should not contain chq_tid - it should be computed after transform", i)

		// Additional validation: chq_sketch field should exist and be []byte
		sketch, sketchOk := row[wkk.NewRowKey("chq_sketch")]
		assert.True(t, sketchOk, "Row %d missing chq_sketch field", i)
		assert.IsType(t, []byte{}, sketch, "Row %d chq_sketch field must be []byte", i)

		// Validate sketch content based on metric type
		sketchBytes := sketch.([]byte)
		// All metrics should now have sketches
		assert.Greater(t, len(sketchBytes), 0, "Row %d (%s) must have non-empty sketch", i, metricType)
	}
}

// Helper function to check if a slice contains a string
func contains(slice []string, item string) bool {
	for _, s := range slice {
		if s == item {
			return true
		}
	}
	return false
}

// TestIngestProtoMetrics_SummarySupport tests that summary data points are properly converted
func TestIngestProtoMetrics_SummarySupport(t *testing.T) {
	// Create test data with summary metrics using pmetric API
	metrics := pmetric.NewMetrics()
	resourceMetrics := metrics.ResourceMetrics().AppendEmpty()
	resourceMetrics.Resource().Attributes().PutStr("service.name", "summary-test-service")

	scopeMetrics := resourceMetrics.ScopeMetrics().AppendEmpty()
	scopeMetrics.Scope().SetName("summary-test-meter")

	// Create a summary metric
	metric := scopeMetrics.Metrics().AppendEmpty()
	metric.SetName("http.request.duration")
	metric.SetDescription("HTTP request duration")
	metric.SetUnit("ms")

	summary := metric.SetEmptySummary()

	// First data point with quantiles
	dp1 := summary.DataPoints().AppendEmpty()
	dp1.Attributes().PutStr("method", "GET")
	dp1.Attributes().PutStr("status", "200")
	dp1.SetStartTimestamp(pcommon.NewTimestampFromTime(time.Unix(1700000000, 0)))
	dp1.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(1700000010, 0)))
	dp1.SetCount(100)
	dp1.SetSum(1500.0)

	// Add quantile values
	qv := dp1.QuantileValues()
	q0 := qv.AppendEmpty()
	q0.SetQuantile(0.0)
	q0.SetValue(5.0)
	q25 := qv.AppendEmpty()
	q25.SetQuantile(0.25)
	q25.SetValue(10.0)
	q50 := qv.AppendEmpty()
	q50.SetQuantile(0.5)
	q50.SetValue(15.0)
	q75 := qv.AppendEmpty()
	q75.SetQuantile(0.75)
	q75.SetValue(20.0)
	q90 := qv.AppendEmpty()
	q90.SetQuantile(0.9)
	q90.SetValue(25.0)
	q95 := qv.AppendEmpty()
	q95.SetQuantile(0.95)
	q95.SetValue(30.0)
	q99 := qv.AppendEmpty()
	q99.SetQuantile(0.99)
	q99.SetValue(50.0)
	q100 := qv.AppendEmpty()
	q100.SetQuantile(1.0)
	q100.SetValue(100.0)

	// Second data point with at least one quantile (otherwise it will be dropped)
	dp2 := summary.DataPoints().AppendEmpty()
	dp2.Attributes().PutStr("method", "POST")
	dp2.Attributes().PutStr("status", "201")
	dp2.SetStartTimestamp(pcommon.NewTimestampFromTime(time.Unix(1700000000, 0)))
	dp2.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(1700000010, 0)))
	dp2.SetCount(50)
	dp2.SetSum(750.0)

	// Add at least one quantile so it won't be dropped
	q2 := dp2.QuantileValues().AppendEmpty()
	q2.SetQuantile(0.5)
	q2.SetValue(15.0)

	// Marshal to protobuf
	marshaler := pmetric.ProtoMarshaler{}
	data, err := marshaler.MarshalMetrics(metrics)
	require.NoError(t, err)

	reader, err := NewIngestProtoMetricsReader(bytes.NewReader(data), ReaderOptions{})
	require.NoError(t, err)
	defer func() { _ = reader.Close() }()

	// Read first batch
	batch, err := reader.Next(context.Background())
	require.NoError(t, err)
	require.NotNil(t, batch)

	// Should have 2 rows for the 2 summary data points
	require.Equal(t, 2, batch.Len())

	// Check first data point (with quantiles)
	row1 := batch.Get(0)
	assert.Equal(t, "http_request_duration", row1[wkk.RowKeyCName])
	assert.Equal(t, "histogram", row1[wkk.RowKeyCMetricType]) // Summaries are treated as histograms

	// Debug: print all keys to see what attributes exist
	// Look for m.method and m.status
	foundMethod := false
	foundStatus := false
	for k, v := range row1 {
		if v == "GET" {
			t.Logf("Found GET at key: %v", k)
			foundMethod = true
		}
		if v == "200" {
			t.Logf("Found 200 at key: %v", k)
			foundStatus = true
		}
	}
	if !foundMethod || !foundStatus {
		t.Logf("Method or status not found with expected values")
	}

	// Check attributes - verify they exist (the keys might be different objects)
	// Since we found the values above, we know the attributes are there
	assert.True(t, foundMethod, "Should have method attribute with value GET")
	assert.True(t, foundStatus, "Should have status attribute with value 200")

	// Check that we have a sketch
	sketch1, ok := row1[wkk.RowKeySketch].([]byte)
	assert.True(t, ok, "First row should have a sketch")
	assert.NotEmpty(t, sketch1, "First row sketch should not be empty")

	// Check rollup values (these will be approximations from the sketch)
	assert.InDelta(t, 1500.0, row1[wkk.RowKeyRollupSum], 150.0, "Sum should be approximately 1500")
	assert.InDelta(t, 100.0, row1[wkk.RowKeyRollupCount], 5.0, "Count should be approximately 100")
	assert.InDelta(t, 15.0, row1[wkk.RowKeyRollupAvg], 2.0, "Average should be approximately 15")

	// Check percentiles (should roughly match the quantile values provided)
	assert.InDelta(t, 10.0, row1[wkk.RowKeyRollupP25], 5.0, "P25 should be approximately 10")
	assert.InDelta(t, 15.0, row1[wkk.RowKeyRollupP50], 5.0, "P50 should be approximately 15")
	assert.InDelta(t, 20.0, row1[wkk.RowKeyRollupP75], 5.0, "P75 should be approximately 20")
	assert.InDelta(t, 30.0, row1[wkk.RowKeyRollupP95], 10.0, "P95 should be approximately 30")
	assert.InDelta(t, 50.0, row1[wkk.RowKeyRollupP99], 15.0, "P99 should be approximately 50")

	// Check second data point (without quantiles, using fallback)
	row2 := batch.Get(1)
	assert.Equal(t, "http_request_duration", row2[wkk.RowKeyCName])

	// Check attributes exist
	foundMethod2 := false
	foundStatus2 := false
	for _, v := range row2 {
		if v == "POST" {
			foundMethod2 = true
		}
		if v == "201" {
			foundStatus2 = true
		}
	}
	assert.True(t, foundMethod2, "Should have method attribute with value POST")
	assert.True(t, foundStatus2, "Should have status attribute with value 201")

	// Check that we have a sketch (created from the single quantile)
	sketch2, ok := row2[wkk.RowKeySketch].([]byte)
	assert.True(t, ok, "Second row should have a sketch")
	assert.NotEmpty(t, sketch2, "Second row sketch should not be empty")

	// The interpolation will create values around the single quantile (15.0)
	assert.Greater(t, row2[wkk.RowKeyRollupCount].(float64), 0.0, "Count should be greater than 0")
	assert.Greater(t, row2[wkk.RowKeyRollupSum].(float64), 0.0, "Sum should be greater than 0")

	// The median should be approximately 15 (the single quantile we provided at 0.5)
	assert.InDelta(t, 15.0, row2[wkk.RowKeyRollupP50], 5.0, "P50 should be approximately 15")

	// Should be EOF on next read
	_, err = reader.Next(context.Background())
	assert.Equal(t, io.EOF, err)
}

// TestIngestProtoMetrics_SummaryEdgeCases tests edge cases for summary data points
func TestIngestProtoMetrics_SummaryEdgeCases(t *testing.T) {
	testCases := []struct {
		name        string
		setupDP     func(dp pmetric.SummaryDataPoint)
		shouldDrop  bool
		description string
	}{
		{
			name: "empty_summary",
			setupDP: func(dp pmetric.SummaryDataPoint) {
				// No count, sum, or quantiles - just timestamp
				dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(1700000010, 0)))
			},
			shouldDrop:  true,
			description: "Summary with no data should be dropped",
		},
		{
			name: "summary_with_single_quantile",
			setupDP: func(dp pmetric.SummaryDataPoint) {
				dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(1700000010, 0)))
				dp.SetCount(10)
				dp.SetSum(100.0)
				qv := dp.QuantileValues().AppendEmpty()
				qv.SetQuantile(0.5)
				qv.SetValue(10.0)
			},
			shouldDrop:  false,
			description: "Summary with single quantile should be processed",
		},
		{
			name: "summary_only_count_and_sum",
			setupDP: func(dp pmetric.SummaryDataPoint) {
				dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(1700000010, 0)))
				dp.SetCount(20)
				dp.SetSum(200.0)
				// No quantiles
			},
			shouldDrop:  true,
			description: "Summary with only count and sum should be dropped (no quantiles)",
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			// Create metrics with pmetric API
			metrics := pmetric.NewMetrics()
			resourceMetrics := metrics.ResourceMetrics().AppendEmpty()
			resourceMetrics.Resource().Attributes().PutStr("service.name", "edge-case-test")

			scopeMetrics := resourceMetrics.ScopeMetrics().AppendEmpty()
			scopeMetrics.Scope().SetName("test-meter")

			metric := scopeMetrics.Metrics().AppendEmpty()
			metric.SetName("test.summary")
			metric.SetDescription("Test summary metric")
			metric.SetUnit("ms")

			summary := metric.SetEmptySummary()
			dp := summary.DataPoints().AppendEmpty()
			tc.setupDP(dp)

			// Marshal to protobuf
			marshaler := pmetric.ProtoMarshaler{}
			data, err := marshaler.MarshalMetrics(metrics)
			require.NoError(t, err)

			reader, err := NewIngestProtoMetricsReader(bytes.NewReader(data), ReaderOptions{})
			require.NoError(t, err)
			defer func() { _ = reader.Close() }()

			batch, err := reader.Next(context.Background())
			if tc.shouldDrop {
				// Should either get empty batch or EOF
				if err != io.EOF {
					require.NoError(t, err)
					assert.Equal(t, 0, batch.Len(), tc.description)
				}
			} else {
				require.NoError(t, err)
				require.NotNil(t, batch)
				assert.Greater(t, batch.Len(), 0, tc.description)

				// Verify we have a sketch
				row := batch.Get(0)
				sketch, ok := row[wkk.RowKeySketch].([]byte)
				assert.True(t, ok, "Row should have a sketch")
				assert.NotEmpty(t, sketch, "Sketch should not be empty")
			}
		})
	}
}
