// 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"
	"testing"
	"time"

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

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

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

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

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

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

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

func TestIngestProtoLogsReader_EmptySlice(t *testing.T) {
	syntheticData := createSyntheticLogData()
	opts := ReaderOptions{
		SignalType: SignalTypeLogs,
		BatchSize:  1,
	}
	reader, err := NewIngestProtoLogsReader(bytes.NewReader(syntheticData), opts)
	require.NoError(t, err)
	defer func() { _ = reader.Close() }()

	// Read with empty slice behavior is no longer applicable with Next() method
	// Next() returns a batch or nil, not dependent on slice size
	batch, err := reader.Next(context.TODO())
	if batch != nil {
		assert.NoError(t, err)
		assert.GreaterOrEqual(t, batch.Len(), 0)
	}
}

func TestIngestProtoLogsReader_Close(t *testing.T) {
	syntheticData := createSyntheticLogData()
	opts := ReaderOptions{
		SignalType: SignalTypeLogs,
		BatchSize:  1,
	}
	reader, err := NewIngestProtoLogsReader(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, "Should read a batch before closing")
	require.Equal(t, 1, batch.Len(), "Should read exactly 1 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)
}

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

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

// Helper function to create synthetic log data using signalbuilder
func createSyntheticLogData() []byte {
	builder := signalbuilder.NewLogBuilder()

	// Define resource attributes
	resourceAttrs := map[string]any{
		"service.name":           "test-log-service",
		"service.version":        "2.0.0",
		"deployment.environment": "test",
	}

	// Create scope logs with different severity levels
	scopeLogs := []signalbuilder.ScopeLogs{
		{
			Name:    "test-logger",
			Version: "1.0.0",
			Attributes: map[string]any{
				"logger.type": "structured",
			},
			LogRecords: []signalbuilder.LogRecord{
				{
					Timestamp:         time.Now().UnixNano(),
					ObservedTimestamp: time.Now().UnixNano(),
					SeverityText:      "DEBUG",
					SeverityNumber:    int32(plog.SeverityNumberDebug),
					Body:              "Debug message for testing",
					Attributes: map[string]any{
						"log.level":    "DEBUG",
						"log.source":   "test-component-0",
						"log.sequence": int64(1),
					},
				},
				{
					Timestamp:         time.Now().Add(time.Second).UnixNano(),
					ObservedTimestamp: time.Now().Add(time.Second).UnixNano(),
					SeverityText:      "INFO",
					SeverityNumber:    int32(plog.SeverityNumberInfo),
					Body:              "Info message with details",
					Attributes: map[string]any{
						"log.level":    "INFO",
						"log.source":   "test-component-1",
						"log.sequence": int64(2),
					},
				},
				{
					Timestamp:         time.Now().Add(2 * time.Second).UnixNano(),
					ObservedTimestamp: time.Now().Add(2 * time.Second).UnixNano(),
					SeverityText:      "WARN",
					SeverityNumber:    int32(plog.SeverityNumberWarn),
					Body:              "Warning about potential issue",
					Attributes: map[string]any{
						"log.level":    "WARN",
						"log.source":   "test-component-2",
						"log.sequence": int64(3),
					},
				},
				{
					Timestamp:         time.Now().Add(3 * time.Second).UnixNano(),
					ObservedTimestamp: time.Now().Add(3 * time.Second).UnixNano(),
					SeverityText:      "ERROR",
					SeverityNumber:    int32(plog.SeverityNumberError),
					Body:              "Error occurred during processing",
					Attributes: map[string]any{
						"log.level":    "ERROR",
						"log.source":   "test-component-3",
						"log.sequence": int64(4),
					},
				},
				{
					Timestamp:         time.Now().Add(4 * time.Second).UnixNano(),
					ObservedTimestamp: time.Now().Add(4 * time.Second).UnixNano(),
					SeverityText:      "FATAL",
					SeverityNumber:    int32(plog.SeverityNumberFatal),
					Body:              "Fatal error - system shutdown",
					Attributes: map[string]any{
						"log.level":    "FATAL",
						"log.source":   "test-component-4",
						"log.sequence": int64(5),
					},
				},
			},
		},
	}

	// Add the resource with scope logs
	resourceLogs := &signalbuilder.ResourceLogs{
		Resource:  resourceAttrs,
		ScopeLogs: scopeLogs,
	}

	err := builder.Add(resourceLogs)
	if err != nil {
		panic(fmt.Sprintf("Failed to add resource logs: %v", err))
	}

	// Build and marshal to protobuf
	logs := builder.Build()
	marshaler := &plog.ProtoMarshaler{}
	data, err := marshaler.MarshalLogs(logs)
	if err != nil {
		panic(fmt.Sprintf("Failed to marshal logs: %v", err))
	}
	return data
}

// Test IngestProtoLogsReader with synthetic log data
func TestIngestProtoLogsReader_SyntheticData(t *testing.T) {
	// Create synthetic log data
	syntheticData := createSyntheticLogData()
	reader := bytes.NewReader(syntheticData)

	opts := ReaderOptions{
		SignalType: SignalTypeLogs,
		BatchSize:  1000,
	}
	protoReader, err := NewIngestProtoLogsReader(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, 5, len(allRows), "Should read exactly 5 log rows from synthetic data")

	// Verify each row has expected fields and values
	expectedSeverities := []string{"DEBUG", "INFO", "WARN", "ERROR", "FATAL"}
	expectedBodies := []string{
		"Debug message for testing",
		"Info message with details",
		"Warning about potential issue",
		"Error occurred during processing",
		"Fatal error - system shutdown",
	}

	for i, row := range allRows {
		t.Run(fmt.Sprintf("log_%d", i), func(t *testing.T) {
			// Should have basic log fields
			_, hasMessage := row[wkk.RowKeyCMessage]
			assert.True(t, hasMessage, "Row should have log body")
			_, hasTimestamp := row[wkk.RowKeyCTimestamp]
			assert.True(t, hasTimestamp, "Row should have timestamp")
			_, hasLevel := row[wkk.NewRowKey("log_level")]
			assert.True(t, hasLevel, "Row should have severity text")

			// Check specific values
			assert.Equal(t, expectedBodies[i], row[wkk.RowKeyCMessage], "Log body should match expected value")
			assert.Equal(t, expectedSeverities[i], row[wkk.NewRowKey("log_level")], "Severity text should match")

			// Check for resource attributes
			_, hasResourceServiceName := row[wkk.NewRowKey("resource_service_name")]
			assert.True(t, hasResourceServiceName, "Should have resource service name")
			assert.Equal(t, "test-log-service", row[wkk.NewRowKey("resource_service_name")])
			_, hasResourceDeploymentEnvironment := row[wkk.NewRowKey("resource_deployment_environment")]
			assert.True(t, hasResourceDeploymentEnvironment, "Should have resource environment")
			assert.Equal(t, "test", row[wkk.NewRowKey("resource_deployment_environment")])

			// Check for scope attributes
			_, hasScopeLoggerType := row[wkk.NewRowKey("scope_logger_type")]
			assert.True(t, hasScopeLoggerType, "Should have scope logger type")
			assert.Equal(t, "structured", row[wkk.NewRowKey("scope_logger_type")])

			// Check for log attributes
			_, hasLogLogLevel := row[wkk.NewRowKey("attr_log_level")]
			assert.True(t, hasLogLogLevel, "Should have log level attribute")
			assert.Equal(t, expectedSeverities[i], row[wkk.NewRowKey("attr_log_level")])
			_, hasLogLogSource := row[wkk.NewRowKey("attr_log_source")]
			assert.True(t, hasLogLogSource, "Should have log source attribute")
			expectedSource := fmt.Sprintf("test-component-%d", i)
			assert.Equal(t, expectedSource, row[wkk.NewRowKey("attr_log_source")])
		})
	}

	// Test batched reading with a new reader instance

	protoReader2, err := NewIngestProtoLogsReader(bytes.NewReader(syntheticData), 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)
				_, hasMessage := row[wkk.RowKeyCMessage]
				assert.True(t, hasMessage)
				_, hasTimestamp := row[wkk.RowKeyCTimestamp]
				assert.True(t, hasTimestamp)
			}
		}

		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

	opts.BatchSize = 1
	protoReader3, err := NewIngestProtoLogsReader(bytes.NewReader(syntheticData), opts)
	require.NoError(t, err)
	defer func() { _ = protoReader3.Close() }()

	batch, err := protoReader3.Next(context.TODO())
	require.NoError(t, err)
	require.NotNil(t, batch, "Should read a batch")
	assert.Equal(t, 1, batch.Len(), "Should read exactly 1 row")
	_, hasMessage := batch.Get(0)[wkk.RowKeyCMessage]
	assert.True(t, hasMessage)
	_, 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 read %d synthetic log rows (expected 5)", len(allRows))
}

// Test IngestProtoLogsReader synthetic data field analysis
func TestIngestProtoLogsReader_SyntheticDataFields(t *testing.T) {
	// Create synthetic log data
	syntheticData := createSyntheticLogData()
	reader := bytes.NewReader(syntheticData)

	opts := ReaderOptions{
		SignalType: SignalTypeLogs,
		BatchSize:  1000,
	}
	protoReader, err := NewIngestProtoLogsReader(reader, opts)
	require.NoError(t, err)
	defer func() { _ = protoReader.Close() }()

	// Read all rows and collect log field info
	allRows, err := readAllRows(protoReader)
	require.NoError(t, err)
	require.Equal(t, 5, len(allRows), "Should read exactly 5 rows for synthetic field analysis")

	severityTexts := make(map[string]int)
	bodyCount := 0

	for _, row := range allRows {
		if body, exists := row[wkk.RowKeyCMessage]; exists {
			if bodyStr, ok := body.(string); ok && bodyStr != "" {
				bodyCount++
			}
		}
		if severityText, exists := row[wkk.NewRowKey("log_level")]; exists {
			if textStr, ok := severityText.(string); ok {
				severityTexts[textStr]++
			}
		}
	}

	t.Logf("Found %d synthetic logs with non-empty bodies", bodyCount)
	t.Logf("Found synthetic severity texts: %+v", severityTexts)

	// Validate expected synthetic data
	assert.Equal(t, 5, bodyCount, "Should have 5 logs with bodies")
	assert.Equal(t, 1, severityTexts["DEBUG"], "Should have 1 DEBUG log")
	assert.Equal(t, 1, severityTexts["INFO"], "Should have 1 INFO log")
	assert.Equal(t, 1, severityTexts["WARN"], "Should have 1 WARN log")
	assert.Equal(t, 1, severityTexts["ERROR"], "Should have 1 ERROR log")
	assert.Equal(t, 1, severityTexts["FATAL"], "Should have 1 FATAL log")
}

// Test IngestProtoLogsReader with structured Go-based synthetic data
func TestIngestProtoLogsReader_SyntheticStructuredData(t *testing.T) {
	builder := signalbuilder.NewLogBuilder()

	// Use Go struct instead of YAML to avoid parsing ambiguity
	resourceLogs := &signalbuilder.ResourceLogs{
		Resource: map[string]any{
			"service.name":           "structured-test-service",
			"service.version":        "3.1.4",
			"deployment.environment": "integration",
			"host.name":              "test-host-01",
		},
		ScopeLogs: []signalbuilder.ScopeLogs{
			{
				Name:    "structured-logger",
				Version: "2.1.0",
				Attributes: map[string]any{
					"logger.framework": "slog",
					"logger.output":    "json",
				},
				LogRecords: []signalbuilder.LogRecord{
					{
						Timestamp:         1640995200000000000,
						ObservedTimestamp: 1640995200100000000,
						SeverityText:      "TRACE",
						SeverityNumber:    int32(plog.SeverityNumberTrace),
						Body:              "Detailed trace information for debugging",
						Attributes: map[string]any{
							"trace_id":    "abc123def456",
							"user_id":     "user-001",
							"operation":   "data_fetch",
							"duration_ms": int64(45),
						},
					},
					{
						Timestamp:         1640995201000000000,
						ObservedTimestamp: 1640995201100000000,
						SeverityText:      "INFO",
						SeverityNumber:    int32(plog.SeverityNumberInfo),
						Body:              "User authentication successful",
						Attributes: map[string]any{
							"user_id":     "user-001",
							"auth_method": "oauth2",
							"session_id":  "sess_987654321",
						},
					},
					{
						Timestamp:         1640995202000000000,
						ObservedTimestamp: 1640995202100000000,
						SeverityText:      "WARN",
						SeverityNumber:    int32(plog.SeverityNumberWarn),
						Body:              "Rate limit approaching for API endpoint",
						Attributes: map[string]any{
							"endpoint":     "/api/v1/data",
							"current_rate": int64(950),
							"rate_limit":   int64(1000),
							"client_ip":    "192.168.1.100",
						},
					},
					{
						Timestamp:         1640995203000000000,
						ObservedTimestamp: 1640995203100000000,
						SeverityText:      "ERROR",
						SeverityNumber:    int32(plog.SeverityNumberError),
						Body:              "Database connection failed",
						Attributes: map[string]any{
							"database":        "primary_db",
							"connection_pool": "pool_1",
							"retry_count":     int64(3),
							"error_code":      "CONNECTION_TIMEOUT",
						},
					},
				},
			},
		},
	}

	err := builder.Add(resourceLogs)
	require.NoError(t, err, "Should successfully add structured resource logs")

	logs := builder.Build()
	marshaler := &plog.ProtoMarshaler{}
	data, err := marshaler.MarshalLogs(logs)
	require.NoError(t, err, "Should successfully marshal logs to protobuf")

	// Test IngestProtoLogsReader with this structured data
	reader := bytes.NewReader(data)
	opts := ReaderOptions{
		SignalType: SignalTypeLogs,
		BatchSize:  1,
	}
	protoReader, err := NewIngestProtoLogsReader(reader, opts)
	require.NoError(t, err)
	require.NotNil(t, protoReader)
	defer func() { _ = protoReader.Close() }()

	// Read all rows from structured data
	allRows, err := readAllRows(protoReader)
	require.NoError(t, err)
	require.Equal(t, 4, len(allRows), "Should read exactly 4 log rows from structured data")

	// Verify structured-specific data
	expectedBodies := []string{
		"Detailed trace information for debugging",
		"User authentication successful",
		"Rate limit approaching for API endpoint",
		"Database connection failed",
	}

	expectedSeverities := []string{"TRACE", "INFO", "WARN", "ERROR"}

	for i, row := range allRows {
		t.Run(fmt.Sprintf("structured_log_%d", i), func(t *testing.T) {
			// Check basic log fields
			assert.Equal(t, expectedBodies[i], row[wkk.RowKeyCMessage], "Structured log body should match")
			assert.Equal(t, expectedSeverities[i], row[wkk.NewRowKey("log_level")], "Structured severity should match")

			// Check resource attributes from structured data
			assert.Equal(t, "structured-test-service", row[wkk.NewRowKey("resource_service_name")])
			assert.Equal(t, "3.1.4", row[wkk.NewRowKey("resource_service_version")])
			assert.Equal(t, "integration", row[wkk.NewRowKey("resource_deployment_environment")])
			assert.Equal(t, "test-host-01", row[wkk.NewRowKey("resource_host_name")])

			// Check scope attributes from structured data
			assert.Equal(t, "slog", row[wkk.NewRowKey("scope_logger_framework")])
			assert.Equal(t, "json", row[wkk.NewRowKey("scope_logger_output")])

			// Check log-specific attributes from structured data with proper types
			switch i {
			case 0: // TRACE log
				assert.Equal(t, "abc123def456", row[wkk.NewRowKey("attr_trace_id")])
				assert.Equal(t, "user-001", row[wkk.NewRowKey("attr_user_id")])
				assert.Equal(t, "data_fetch", row[wkk.NewRowKey("attr_operation")])
				// OTEL attribute processing may convert numbers to strings regardless of input type
				t.Logf("duration_ms type: %T, value: %v", row[wkk.NewRowKey("attr_duration_ms")], row[wkk.NewRowKey("attr_duration_ms")])
				assert.Contains(t, []any{int64(45), "45"}, row[wkk.NewRowKey("attr_duration_ms")])
			case 1: // INFO log
				assert.Equal(t, "user-001", row[wkk.NewRowKey("attr_user_id")])
				assert.Equal(t, "oauth2", row[wkk.NewRowKey("attr_auth_method")])
				assert.Equal(t, "sess_987654321", row[wkk.NewRowKey("attr_session_id")])
			case 2: // WARN log
				assert.Equal(t, "/api/v1/data", row[wkk.NewRowKey("attr_endpoint")])
				// OTEL attribute processing may convert numbers to strings regardless of input type
				assert.Contains(t, []any{int64(950), "950"}, row[wkk.NewRowKey("attr_current_rate")])
				assert.Contains(t, []any{int64(1000), "1000"}, row[wkk.NewRowKey("attr_rate_limit")])
				assert.Equal(t, "192.168.1.100", row[wkk.NewRowKey("attr_client_ip")])
			case 3: // ERROR log
				assert.Equal(t, "primary_db", row[wkk.NewRowKey("attr_database")])
				assert.Equal(t, "pool_1", row[wkk.NewRowKey("attr_connection_pool")])
				// OTEL attribute processing may convert numbers to strings regardless of input type
				assert.Contains(t, []any{int64(3), "3"}, row[wkk.NewRowKey("attr_retry_count")])
				assert.Equal(t, "CONNECTION_TIMEOUT", row[wkk.NewRowKey("attr_error_code")])
			}
		})
	}

	t.Logf("Successfully read %d structured log rows", len(allRows))
}

// Test IngestProtoLogsReader with multi-resource synthetic data
func TestIngestProtoLogsReader_MultiResourceSyntheticData(t *testing.T) {
	builder := signalbuilder.NewLogBuilder()

	// Add logs from multiple services/resources
	resources := []struct {
		name   string
		attrs  map[string]any
		scopes []signalbuilder.ScopeLogs
	}{
		{
			name: "web-service",
			attrs: map[string]any{
				"service.name":    "web-frontend",
				"service.version": "1.2.3",
				"environment":     "production",
			},
			scopes: []signalbuilder.ScopeLogs{
				{
					Name:    "http-logger",
					Version: "1.0.0",
					LogRecords: []signalbuilder.LogRecord{
						{
							Timestamp:      time.Now().UnixNano(),
							SeverityText:   "INFO",
							SeverityNumber: int32(plog.SeverityNumberInfo),
							Body:           "HTTP request processed",
							Attributes:     map[string]any{"method": "GET", "status": 200, "path": "/api/users"},
						},
						{
							Timestamp:      time.Now().Add(time.Second).UnixNano(),
							SeverityText:   "ERROR",
							SeverityNumber: int32(plog.SeverityNumberError),
							Body:           "HTTP request failed",
							Attributes:     map[string]any{"method": "POST", "status": 500, "path": "/api/orders"},
						},
					},
				},
			},
		},
		{
			name: "database-service",
			attrs: map[string]any{
				"service.name":    "postgres-db",
				"service.version": "13.7",
				"environment":     "production",
			},
			scopes: []signalbuilder.ScopeLogs{
				{
					Name:    "db-logger",
					Version: "2.0.0",
					LogRecords: []signalbuilder.LogRecord{
						{
							Timestamp:      time.Now().Add(2 * time.Second).UnixNano(),
							SeverityText:   "DEBUG",
							SeverityNumber: int32(plog.SeverityNumberDebug),
							Body:           "Query executed successfully",
							Attributes:     map[string]any{"query": "SELECT * FROM users", "duration": "15ms"},
						},
						{
							Timestamp:      time.Now().Add(3 * time.Second).UnixNano(),
							SeverityText:   "WARN",
							SeverityNumber: int32(plog.SeverityNumberWarn),
							Body:           "Slow query detected",
							Attributes:     map[string]any{"query": "SELECT * FROM orders", "duration": "2500ms"},
						},
					},
				},
			},
		},
	}

	// Add each resource to the builder
	for _, res := range resources {
		resourceLogs := &signalbuilder.ResourceLogs{
			Resource:  res.attrs,
			ScopeLogs: res.scopes,
		}
		err := builder.Add(resourceLogs)
		require.NoError(t, err, "Should add resource logs for %s", res.name)
	}

	// Build and test
	logs := builder.Build()
	marshaler := &plog.ProtoMarshaler{}
	data, err := marshaler.MarshalLogs(logs)
	require.NoError(t, err)

	reader := bytes.NewReader(data)

	opts := ReaderOptions{
		SignalType: SignalTypeLogs,
		BatchSize:  1,
	}
	protoReader, err := NewIngestProtoLogsReader(reader, opts)
	require.NoError(t, err)
	defer func() { _ = protoReader.Close() }()

	// Should read logs from both resources (4 total)
	allRows, err := readAllRows(protoReader)
	require.NoError(t, err)
	require.Equal(t, 4, len(allRows), "Should read logs from both services")

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

	assert.Equal(t, 2, webLogs, "Should have 2 logs from web service")
	assert.Equal(t, 2, dbLogs, "Should have 2 logs from database service")

	t.Logf("Successfully read %d logs from %d services", len(allRows), len(resources))
}
