// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//	http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package stream

import (
	"fmt"
	"testing"

	"github.com/stretchr/testify/assert"

	"github.com/apache/skywalking-banyandb/pkg/bytes"
	"github.com/apache/skywalking-banyandb/pkg/convert"
	"github.com/apache/skywalking-banyandb/pkg/encoding"
	pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)

func TestTag_reset(t *testing.T) {
	tt := &tag{
		name:      "test",
		valueType: pbv1.ValueTypeStr,
		values:    [][]byte{[]byte("value1"), []byte("value2")},
	}

	tt.reset()

	assert.Equal(t, "", tt.name)
	assert.Equal(t, 0, len(tt.values))
}

func TestTag_resizeValues(t *testing.T) {
	tt := &tag{
		values: make([][]byte, 2, 5),
	}

	values := tt.resizeValues(3)
	assert.Equal(t, 3, len(values))
	assert.Equal(t, 5, cap(values))

	values = tt.resizeValues(6)
	assert.Equal(t, 6, len(values))
	assert.True(t, cap(values) >= 6) // The capacity is at least 6, but could be more
}

func TestTag_mustWriteTo_mustReadValues(t *testing.T) {
	tests := []struct {
		tag  *tag
		name string
	}{
		{
			name: "string with nils",
			tag: &tag{
				name:      "test",
				valueType: pbv1.ValueTypeStr,
				values:    [][]byte{[]byte("value1"), nil, []byte("value2"), nil},
			},
		},
		{
			name: "int64 with null",
			tag: &tag{
				name:      "test",
				valueType: pbv1.ValueTypeInt64,
				values:    [][]byte{[]byte("null"), nil, []byte("null"), nil},
			},
		},
		{
			name: "valid int64 values",
			tag: &tag{
				name:      "test",
				valueType: pbv1.ValueTypeInt64,
				values: [][]byte{
					convert.Int64ToBytes(1),
					convert.Int64ToBytes(2),
					convert.Int64ToBytes(4),
					convert.Int64ToBytes(5),
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tm := &tagMetadata{}

			buf, filterBuf := &bytes.Buffer{}, &bytes.Buffer{}
			w, fw := &writer{}, &writer{}
			w.init(buf)
			fw.init(filterBuf)

			tt.tag.mustWriteTo(tm, w, fw)
			assert.Equal(t, w.bytesWritten, tm.size)
			assert.Equal(t, uint64(len(buf.Buf)), tm.size)
			assert.Equal(t, uint64(0), tm.offset)
			assert.Equal(t, tt.tag.name, tm.name)
			assert.Equal(t, tt.tag.valueType, tm.valueType)

			decoder := &encoding.BytesBlockDecoder{}
			unmarshaled := &tag{}
			unmarshaled.mustReadValues(decoder, buf, *tm, uint64(len(tt.tag.values)))

			assert.Equal(t, tt.tag.name, unmarshaled.name)
			assert.Equal(t, tt.tag.valueType, unmarshaled.valueType)
			assert.Equal(t, tt.tag.values, unmarshaled.values)
		})
	}
}

func TestTagFamily_reset(t *testing.T) {
	tf := &tagFamily{
		name: "test",
		tags: []tag{
			{
				name:      "test1",
				valueType: pbv1.ValueTypeStr,
				values:    [][]byte{[]byte("value1"), []byte("value2")},
			},
			{
				name:      "test2",
				valueType: pbv1.ValueTypeInt64,
				values:    [][]byte{[]byte("value3"), []byte("value4")},
			},
		},
	}

	tf.reset()

	assert.Equal(t, "", tf.name)
	assert.Equal(t, 0, len(tf.tags))
}

func TestTagFamily_resizeTags(t *testing.T) {
	tf := &tagFamily{
		tags: make([]tag, 2, 5),
	}

	tags := tf.resizeTags(3)
	assert.Equal(t, 3, len(tags))
	assert.Equal(t, 5, cap(tags))

	tags = tf.resizeTags(6)
	assert.Equal(t, 6, len(tags))
	assert.True(t, cap(tags) >= 6) // The capacity is at least 6, but could be more
}

func TestTag_HighCardinalityStringEncoding(t *testing.T) {
	tests := []struct {
		name            string
		description     string
		expectedEncType encoding.EncodeType
		uniqueCount     int
		totalCount      int
	}{
		{
			name:            "exactly 256 unique values - should use dictionary",
			uniqueCount:     256,
			totalCount:      256,
			expectedEncType: encoding.EncodeTypeDictionary,
			description:     "Dictionary encoding should be used when exactly at the threshold",
		},
		{
			name:            "257 unique values - should use plain encoding",
			uniqueCount:     257,
			totalCount:      257,
			expectedEncType: encoding.EncodeTypePlain,
			description:     "Plain encoding should be used when exceeding dictionary threshold",
		},
		{
			name:            "300 unique values - should use plain encoding",
			uniqueCount:     300,
			totalCount:      300,
			expectedEncType: encoding.EncodeTypePlain,
			description:     "Plain encoding should be used for high cardinality strings",
		},
		{
			name:            "1000 unique values - should use plain encoding",
			uniqueCount:     1000,
			totalCount:      1000,
			expectedEncType: encoding.EncodeTypePlain,
			description:     "Plain encoding should be used for very high cardinality",
		},
		{
			name:            "500 total with 200 unique - should use dictionary",
			uniqueCount:     200,
			totalCount:      500,
			expectedEncType: encoding.EncodeTypeDictionary,
			description:     "Dictionary should be used when unique count is below threshold despite high total count",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Generate unique string values
			values := make([][]byte, tt.totalCount)

			// Create unique values up to uniqueCount
			for i := 0; i < tt.uniqueCount; i++ {
				values[i] = []byte(fmt.Sprintf("unique_value_%06d", i))
			}

			// If totalCount > uniqueCount, repeat some values to reach totalCount
			for i := tt.uniqueCount; i < tt.totalCount; i++ {
				// Repeat values cyclically
				repeatIndex := i % tt.uniqueCount
				values[i] = []byte(fmt.Sprintf("unique_value_%06d", repeatIndex))
			}

			testTag := &tag{
				name:      "high_cardinality_tag",
				valueType: pbv1.ValueTypeStr,
				values:    values,
			}

			// Encode the tag
			tm := &tagMetadata{}
			buf, filterBuf := &bytes.Buffer{}, &bytes.Buffer{}
			w, fw := &writer{}, &writer{}
			w.init(buf)
			fw.init(filterBuf)

			testTag.mustWriteTo(tm, w, fw)

			// Verify basic metadata
			assert.Equal(t, w.bytesWritten, tm.size)
			assert.Equal(t, uint64(len(buf.Buf)), tm.size)
			assert.Equal(t, uint64(0), tm.offset)
			assert.Equal(t, testTag.name, tm.name)
			assert.Equal(t, testTag.valueType, tm.valueType)

			// The new shared encoding uses automatic compression, so we can't directly check the encoding type
			// from the first byte anymore. Instead, we verify the round-trip works correctly.
			assert.True(t, len(buf.Buf) > 0, "Encoded buffer should not be empty")

			// Test roundtrip: decode and verify all values are preserved
			decoder := &encoding.BytesBlockDecoder{}
			unmarshaled := &tag{}
			unmarshaled.mustReadValues(decoder, buf, *tm, uint64(len(testTag.values)))

			assert.Equal(t, testTag.name, unmarshaled.name)
			assert.Equal(t, testTag.valueType, unmarshaled.valueType)
			assert.Equal(t, len(testTag.values), len(unmarshaled.values), "Number of values should match")

			// Verify all values are correctly decoded
			for i, originalValue := range testTag.values {
				assert.Equal(t, originalValue, unmarshaled.values[i],
					"Value at index %d should match original", i)
			}
		})
	}
}

func TestEncodeTypedTag(t *testing.T) {
	assert.Equal(t, "state#int", encodeTypedTag("state", pbv1.ValueTypeInt64))
	assert.Equal(t, "state#str", encodeTypedTag("state", pbv1.ValueTypeStr))
	assert.Equal(t, "http.status#int", encodeTypedTag("http.status", pbv1.ValueTypeInt64))
	assert.Equal(t, "data#bin", encodeTypedTag("data", pbv1.ValueTypeBinaryData))
	assert.Equal(t, "unknown", encodeTypedTag("unknown", pbv1.ValueTypeUnknown))
}

func TestDecodeTypedTag(t *testing.T) {
	assert.Equal(t, "state", decodeTypedTag("state#int"))
	assert.Equal(t, "state", decodeTypedTag("state#str"))
	assert.Equal(t, "http.status", decodeTypedTag("http.status#int"))
	assert.Equal(t, "http.status", decodeTypedTag("http.status"))
	assert.Equal(t, "simple", decodeTypedTag("simple"))
	assert.Equal(t, "my.str", decodeTypedTag("my.str"))
	assert.Equal(t, "arr.int", decodeTypedTag("arr.int"))
	assert.Equal(t, "data.bin", decodeTypedTag("data.bin"))
}
