// 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 inverted

import (
	"context"
	"maps"
	"sort"
	"testing"
	"time"

	"github.com/blugelabs/bluge"
	"github.com/blugelabs/bluge/numeric"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
	modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
	"github.com/apache/skywalking-banyandb/pkg/convert"
	"github.com/apache/skywalking-banyandb/pkg/index"
	"github.com/apache/skywalking-banyandb/pkg/logger"
	"github.com/apache/skywalking-banyandb/pkg/timestamp"
)

var (
	fieldKeyDuration = index.FieldKey{
		IndexRuleID: indexRuleID,
	}
	fieldKeyServiceName = index.FieldKey{
		IndexRuleID: 6,
	}
	fieldKeyStartTime = index.FieldKey{
		IndexRuleID: 21,
	}
)

func TestStore_Search(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()

	// Setup some data
	insertData(tester, s)

	// Test cases
	tests := []struct {
		term       [][]byte
		want       []index.SeriesDocument
		projection []index.FieldKey
	}{
		{
			term: [][]byte{[]byte("test1")},
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test1"),
					},
				},
			},
		},
		{
			term:       [][]byte{[]byte("test1"), []byte("test2"), []byte("test3"), []byte("foo")},
			projection: []index.FieldKey{fieldKeyDuration, fieldKeyServiceName},
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal():    convert.Int64ToBytes(int64(100)),
						fieldKeyServiceName.Marshal(): []byte("svc2"),
					},
					Timestamp: int64(101),
				},
				{
					Key: index.Series{
						EntityValues: []byte("test1"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal():    nil,
						fieldKeyServiceName.Marshal(): nil,
					},
				},
				{
					Key: index.Series{
						EntityValues: []byte("test3"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal():    convert.Int64ToBytes(int64(500)),
						fieldKeyServiceName.Marshal(): nil,
					},
					Timestamp: int64(1001),
				},
			},
		},
		{
			term:       [][]byte{[]byte("test1"), []byte("test2"), []byte("test3"), []byte("foo")},
			projection: []index.FieldKey{fieldKeyDuration},
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal(): convert.Int64ToBytes(int64(100)),
					},
					Timestamp: int64(101),
				},

				{
					Key: index.Series{
						EntityValues: []byte("test1"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal(): nil,
					},
				},
				{
					Key: index.Series{
						EntityValues: []byte("test3"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal(): convert.Int64ToBytes(int64(500)),
					},
					Timestamp: int64(1001),
				},
			},
		},
		{
			term:       [][]byte{[]byte("test1"), []byte("test2"), []byte("test3"), []byte("foo")},
			projection: []index.FieldKey{fieldKeyServiceName},
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Fields: map[string][]byte{
						fieldKeyServiceName.Marshal(): []byte("svc2"),
					},
					Timestamp: int64(101),
				},
				{
					Key: index.Series{
						EntityValues: []byte("test1"),
					},
					Fields: map[string][]byte{
						fieldKeyServiceName.Marshal(): nil,
					},
				},
				{
					Key: index.Series{
						EntityValues: []byte("test3"),
					},
					Fields: map[string][]byte{
						fieldKeyServiceName.Marshal(): nil,
					},
					Timestamp: int64(1001),
				},
			},
		},
		{
			term: [][]byte{[]byte("foo")},
			want: emptySeries,
		},
	}

	for _, tt := range tests {
		var matchers []index.SeriesMatcher
		var name string
		for _, term := range tt.term {
			matchers = append(matchers, index.SeriesMatcher{
				Type:  index.SeriesMatcherTypeExact,
				Match: term,
			})
			name += string(term) + "-"
		}
		t.Run(name, func(t *testing.T) {
			query, err := s.BuildQuery(matchers, nil, nil)
			require.NotEmpty(t, query.String())
			require.NoError(t, err)
			got, err := s.Search(context.Background(), tt.projection, query, 0)
			require.NoError(t, err)
			sort.Slice(tt.want, func(i, j int) bool {
				return string(tt.want[i].Key.EntityValues) < string(tt.want[j].Key.EntityValues)
			})
			sort.Slice(got, func(i, j int) bool {
				return string(got[i].Key.EntityValues) < string(got[j].Key.EntityValues)
			})
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestStore_SearchWildcard(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()

	// Setup some data
	insertData(tester, s)

	// Test cases
	tests := []struct {
		wildcard   []byte
		projection []index.FieldKey
		want       []index.SeriesDocument
	}{
		{
			wildcard: []byte("test*"),
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test1"),
					},
				},
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Timestamp: int64(101),
				},
				{
					Key: index.Series{
						EntityValues: []byte("test3"),
					},
					Timestamp: int64(1001),
				},
				{
					Key: index.Series{
						EntityValues: []byte("test4"),
					},
					Timestamp: int64(2001),
				},
			},
		},
		{
			wildcard: []byte("*2"),
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Timestamp: int64(101),
				},
			},
		},
		{
			wildcard: []byte("t*st1"),
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test1"),
					},
				},
			},
		},
		{
			wildcard: []byte("foo*"),
			want:     []index.SeriesDocument{},
		},
	}

	for _, tt := range tests {
		t.Run(string(tt.wildcard), func(t *testing.T) {
			query, err := s.BuildQuery([]index.SeriesMatcher{
				{
					Type:  index.SeriesMatcherTypeWildcard,
					Match: tt.wildcard,
				},
			}, nil, nil)
			require.NoError(t, err)
			require.NotEmpty(t, query.String())
			got, err := s.Search(context.Background(), tt.projection, query, 0)
			require.NoError(t, err)
			assert.ElementsMatch(t, tt.want, got)
		})
	}
}

func TestStore_SearchPrefix(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()

	// Setup some data
	insertData(tester, s)

	// Test cases
	tests := []struct {
		prefix     []byte
		projection []index.FieldKey
		want       []index.SeriesDocument
	}{
		{
			prefix: []byte("test"),
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test1"),
					},
				},
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Timestamp: int64(101),
				},
				{
					Key: index.Series{
						EntityValues: []byte("test3"),
					},
					Timestamp: int64(1001),
				},
				{
					Key: index.Series{
						EntityValues: []byte("test4"),
					},
					Timestamp: int64(2001),
				},
			},
		},
		{
			prefix: []byte("foo"),
			want:   []index.SeriesDocument{},
		},
	}

	for _, tt := range tests {
		t.Run(string(tt.prefix), func(t *testing.T) {
			query, err := s.BuildQuery([]index.SeriesMatcher{
				{
					Type:  index.SeriesMatcherTypePrefix,
					Match: tt.prefix,
				},
			}, nil, nil)
			require.NoError(t, err)
			require.NotEmpty(t, query.String())
			got, err := s.Search(context.Background(), tt.projection, query, 0)
			require.NoError(t, err)
			assert.ElementsMatch(t, tt.want, got)
		})
	}
}

func TestStore_SearchWithSecondaryQuery(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()

	// Setup some data
	insertData(tester, s)

	// Define the secondary query
	secondaryQuery := &queryNode{
		query: bluge.NewTermQuery("svc2").SetField(fieldKeyServiceName.Marshal()),
		node:  newTermNode("svc2", nil),
	}

	// Test cases
	tests := []struct {
		term       [][]byte
		want       []index.SeriesDocument
		projection []index.FieldKey
	}{
		{
			term:       [][]byte{[]byte("test2")},
			projection: []index.FieldKey{fieldKeyServiceName, fieldKeyDuration, {TagName: "short_name"}},
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal():    convert.Int64ToBytes(int64(100)),
						fieldKeyServiceName.Marshal(): []byte("svc2"),
						"short_name":                  []byte("t2"),
					},
					Timestamp: int64(101),
				},
			},
		},
		{
			term:       [][]byte{[]byte("test3")},
			projection: []index.FieldKey{fieldKeyServiceName, fieldKeyDuration},
			want:       []index.SeriesDocument{},
		},
		{
			term:       [][]byte{[]byte("test1")},
			projection: []index.FieldKey{fieldKeyServiceName, fieldKeyDuration},
			want:       []index.SeriesDocument{},
		},
		{
			term:       [][]byte{[]byte("test2"), []byte("test3")},
			projection: []index.FieldKey{fieldKeyServiceName, fieldKeyDuration, {TagName: "short_name"}},
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal():    convert.Int64ToBytes(int64(100)),
						fieldKeyServiceName.Marshal(): []byte("svc2"),
						"short_name":                  []byte("t2"),
					},
					Timestamp: int64(101),
				},
			},
		},
		{
			term:       [][]byte{[]byte("test1"), []byte("test2")},
			projection: []index.FieldKey{fieldKeyServiceName, fieldKeyDuration},
			want: []index.SeriesDocument{
				{
					Key: index.Series{
						EntityValues: []byte("test2"),
					},
					Fields: map[string][]byte{
						fieldKeyDuration.Marshal():    convert.Int64ToBytes(int64(100)),
						fieldKeyServiceName.Marshal(): []byte("svc2"),
					},
					Timestamp: int64(101),
				},
			},
		},
	}

	for _, tt := range tests {
		var matchers []index.SeriesMatcher
		var name string
		for _, term := range tt.term {
			matchers = append(matchers, index.SeriesMatcher{
				Type:  index.SeriesMatcherTypeExact,
				Match: term,
			})
			name += string(term) + "-"
		}
		t.Run(name, func(t *testing.T) {
			query, err := s.BuildQuery(matchers, secondaryQuery, nil)
			require.NotEmpty(t, query.String())
			require.NoError(t, err)
			got, err := s.Search(context.Background(), tt.projection, query, 0)
			require.NoError(t, err)
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestStore_SeriesSort(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()

	// Setup some data
	updateData(tester, s)

	// Define the order by field
	orderBy := &index.OrderBy{
		Index: &databasev1.IndexRule{
			Metadata: &commonv1.Metadata{
				Id: fieldKeyStartTime.IndexRuleID,
			},
		},
		Sort: modelv1.Sort_SORT_ASC,
		Type: index.OrderByTypeIndex,
	}

	// Test cases
	tests := []struct {
		name      string
		orderBy   *index.OrderBy
		timeRange *timestamp.TimeRange
		want      []index.DocumentResult
		fieldKeys []index.FieldKey
	}{
		{
			name:      "Sort by start_time ascending",
			orderBy:   orderBy,
			fieldKeys: []index.FieldKey{fieldKeyStartTime},
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test2"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(100)),
					},
					SortedValue: convert.Int64ToBytes(int64(100)),
				},
				{
					EntityValues: []byte("test3"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(1000)),
					},
					SortedValue: convert.Int64ToBytes(int64(1000)),
				},
				{
					EntityValues: []byte("test4"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(2000)),
					},
					SortedValue: convert.Int64ToBytes(int64(2000)),
				},
				{
					EntityValues: []byte("test1"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): nil,
					},
					SortedValue: []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff},
				},
			},
		},
		{
			name: "Sort by start_time descending",
			orderBy: &index.OrderBy{
				Index: &databasev1.IndexRule{
					Metadata: &commonv1.Metadata{
						Id: fieldKeyStartTime.IndexRuleID,
					},
				},
				Sort: modelv1.Sort_SORT_DESC,
				Type: index.OrderByTypeIndex,
			},
			fieldKeys: []index.FieldKey{fieldKeyStartTime},
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test4"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(2000)),
					},
					SortedValue: convert.Int64ToBytes(int64(2000)),
				},
				{
					EntityValues: []byte("test3"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(1000)),
					},
					SortedValue: convert.Int64ToBytes(int64(1000)),
				},
				{
					EntityValues: []byte("test2"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(100)),
					},
					SortedValue: convert.Int64ToBytes(int64(100)),
				},
				{
					EntityValues: []byte("test1"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): nil,
					},
					SortedValue: []byte{0x00},
				},
			},
		},
		{
			name:      "Sort by start_time ascending with time range 100 to 1000",
			orderBy:   orderBy,
			fieldKeys: []index.FieldKey{fieldKeyStartTime},
			timeRange: func() *timestamp.TimeRange {
				tr := timestamp.NewInclusiveTimeRange(time.Unix(0, 100), time.Unix(0, 1000))
				return &tr
			}(),
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test2"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(100)),
					},
					SortedValue: convert.Int64ToBytes(int64(100)),
				},
				{
					EntityValues: []byte("test3"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(1000)),
					},
					SortedValue: convert.Int64ToBytes(int64(1000)),
				},
			},
		},
		{
			name:      "Sort by start_time ascending with time range 0 to 2000",
			orderBy:   orderBy,
			fieldKeys: []index.FieldKey{fieldKeyStartTime},
			timeRange: func() *timestamp.TimeRange {
				tr := timestamp.NewInclusiveTimeRange(time.Unix(0, 0), time.Unix(0, 2000))
				return &tr
			}(),
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test2"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(100)),
					},
					SortedValue: convert.Int64ToBytes(int64(100)),
				},
				{
					EntityValues: []byte("test3"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(1000)),
					},
					SortedValue: convert.Int64ToBytes(int64(1000)),
				},
				{
					EntityValues: []byte("test4"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(2000)),
					},
					SortedValue: convert.Int64ToBytes(int64(2000)),
				},
			},
		},
		{
			name:      "Sort by start_time ascending with time range 500 to 1500",
			orderBy:   orderBy,
			fieldKeys: []index.FieldKey{fieldKeyStartTime},
			timeRange: func() *timestamp.TimeRange {
				tr := timestamp.NewInclusiveTimeRange(time.Unix(0, 500), time.Unix(0, 1500))
				return &tr
			}(),
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test3"),
					Values: map[string][]byte{
						fieldKeyStartTime.Marshal(): convert.Int64ToBytes(int64(1000)),
					},
					SortedValue: convert.Int64ToBytes(int64(1000)),
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var secondaryQuery index.Query
			if tt.timeRange != nil {
				secondaryQuery = &queryNode{
					query: bluge.NewTermRangeInclusiveQuery(
						string(convert.Int64ToBytes(tt.timeRange.Start.Local().UnixNano())),
						string(convert.Int64ToBytes(tt.timeRange.End.Local().UnixNano())),
						tt.timeRange.IncludeStart,
						tt.timeRange.IncludeEnd,
					).SetField(fieldKeyStartTime.Marshal()),
				}
			}
			query, err := s.BuildQuery([]index.SeriesMatcher{
				{
					Type:  index.SeriesMatcherTypePrefix,
					Match: []byte("test"),
				},
			}, secondaryQuery, nil)
			require.NoError(t, err)
			iter, err := s.SeriesSort(context.Background(), query, tt.orderBy, 10, tt.fieldKeys)
			require.NoError(t, err)
			defer iter.Close()

			var got []index.DocumentResult
			for iter.Next() {
				var g index.DocumentResult
				val := iter.Val()
				g.DocID = val.DocID
				g.EntityValues = val.EntityValues
				g.Values = maps.Clone(val.Values)
				g.SortedValue = val.SortedValue
				got = append(got, g)
			}
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestStore_TimestampSort(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()

	// Setup some data
	updateData(tester, s)

	// Define the order by field
	orderBy := &index.OrderBy{
		Type: index.OrderByTypeTime,
		Sort: modelv1.Sort_SORT_ASC,
	}

	// Test cases
	tests := []struct {
		name      string
		orderBy   *index.OrderBy
		timeRange *timestamp.TimeRange
		want      []index.DocumentResult
		fieldKeys []index.FieldKey
	}{
		{
			name:    "Sort by timestamp ascending",
			orderBy: orderBy,
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test2"),
					Timestamp:    int64(101),
					SortedValue:  numeric.MustNewPrefixCodedInt64(101, 0),
				},
				{
					EntityValues: []byte("test3"),
					Timestamp:    int64(1001),
					SortedValue:  numeric.MustNewPrefixCodedInt64(1001, 0),
				},
				{
					EntityValues: []byte("test4"),
					Timestamp:    int64(2001),
					SortedValue:  numeric.MustNewPrefixCodedInt64(2001, 0),
				},
				{
					EntityValues: []byte("test1"),
					Timestamp:    0,
					SortedValue:  []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff},
				},
			},
		},
		{
			name: "Sort by timestamp descending",
			orderBy: &index.OrderBy{
				Type: index.OrderByTypeTime,
				Sort: modelv1.Sort_SORT_DESC,
			},
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test4"),
					Timestamp:    int64(2001),
					SortedValue:  numeric.MustNewPrefixCodedInt64(2001, 0),
				},
				{
					EntityValues: []byte("test3"),
					Timestamp:    int64(1001),
					SortedValue:  numeric.MustNewPrefixCodedInt64(1001, 0),
				},
				{
					EntityValues: []byte("test2"),
					Timestamp:    int64(101),
					SortedValue:  numeric.MustNewPrefixCodedInt64(101, 0),
				},
				{
					EntityValues: []byte("test1"),
					Timestamp:    0,
					SortedValue:  []byte{0x00},
				},
			},
		},
		{
			name:    "Sort by timestamp ascending with time range 100 to 1000",
			orderBy: orderBy,
			timeRange: func() *timestamp.TimeRange {
				tr := timestamp.NewInclusiveTimeRange(time.Unix(0, 100), time.Unix(0, 1000))
				return &tr
			}(),
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test2"),
					Timestamp:    int64(101),
					SortedValue:  numeric.MustNewPrefixCodedInt64(101, 0),
				},
			},
		},
		{
			name:    "Sort by timestamp ascending with time range 0 to 2000",
			orderBy: orderBy,
			timeRange: func() *timestamp.TimeRange {
				tr := timestamp.NewInclusiveTimeRange(time.Unix(0, 0), time.Unix(0, 2000))
				return &tr
			}(),
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test2"),
					Timestamp:    int64(101),
					SortedValue:  numeric.MustNewPrefixCodedInt64(101, 0),
				},
				{
					EntityValues: []byte("test3"),
					Timestamp:    int64(1001),
					SortedValue:  numeric.MustNewPrefixCodedInt64(1001, 0),
				},
			},
		},
		{
			name:    "Sort by timestamp ascending with time range 500 to 1500",
			orderBy: orderBy,
			timeRange: func() *timestamp.TimeRange {
				tr := timestamp.NewInclusiveTimeRange(time.Unix(0, 500), time.Unix(0, 1500))
				return &tr
			}(),
			want: []index.DocumentResult{
				{
					EntityValues: []byte("test3"),
					Timestamp:    int64(1001),
					SortedValue:  numeric.MustNewPrefixCodedInt64(1001, 0),
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			query, err := s.BuildQuery([]index.SeriesMatcher{
				{
					Type:  index.SeriesMatcherTypePrefix,
					Match: []byte("test"),
				},
			}, nil, tt.timeRange)
			require.NoError(t, err)
			iter, err := s.SeriesSort(context.Background(), query, tt.orderBy, 10, tt.fieldKeys)
			require.NoError(t, err)
			defer iter.Close()

			var got []index.DocumentResult
			for iter.Next() {
				var g index.DocumentResult
				val := iter.Val()
				g.DocID = val.DocID
				g.EntityValues = val.EntityValues
				if len(val.Values) > 0 {
					g.Values = maps.Clone(val.Values)
				}
				g.SortedValue = val.SortedValue
				g.Timestamp = val.Timestamp
				got = append(got, g)
			}
			assert.Equal(t, tt.want, got)
		})
	}
}

func insertData(tester *require.Assertions, s index.SeriesStore) {
	b1, b2 := generateDocs()
	tester.NoError(s.InsertSeriesBatch(b1))
	tester.NoError(s.InsertSeriesBatch(b2))
}

func generateDocs() (index.Batch, index.Batch) {
	series1 := index.Document{
		EntityValues: []byte("test1"),
	}

	series2 := index.Document{
		EntityValues: []byte("test2"),
		Fields: []index.Field{
			field(fieldKeyDuration, convert.Int64ToBytes(100), true),
			field(fieldKeyServiceName, []byte("svc2"), true),
			field(fieldKeyStartTime, convert.Int64ToBytes(100), true),
			field(index.FieldKey{TagName: "short_name"}, []byte("t2"), false),
		},
		Timestamp: int64(101),
	}

	series3 := index.Document{
		EntityValues: []byte("test3"),
		Fields: []index.Field{
			field(fieldKeyDuration, convert.Int64ToBytes(500), true),
			field(fieldKeyStartTime, convert.Int64ToBytes(1000), true),
			field(index.FieldKey{TagName: "short_name"}, []byte("t3"), false),
		},
		Timestamp: int64(1001),
	}
	series4 := index.Document{
		EntityValues: []byte("test4"),
		Fields: []index.Field{
			field(fieldKeyDuration, convert.Int64ToBytes(500), true),
			field(fieldKeyStartTime, convert.Int64ToBytes(2000), true),
		},
		Timestamp: int64(2001),
	}
	return index.Batch{
			Documents: []index.Document{series1, series2, series4, series3},
		}, index.Batch{
			Documents: []index.Document{series3},
		}
}

func field(key index.FieldKey, value []byte, indexed bool) index.Field {
	f := index.NewBytesField(key, value)
	f.Index = indexed
	f.Store = true
	return f
}

// TestStore_BuildQueryTimeRangeWithoutSeriesMatchers verifies that BuildQuery applies the time range
// filter even when no series matchers are provided, both with and without a secondary query.
func TestStore_BuildQueryTimeRangeWithoutSeriesMatchers(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()

	insertData(tester, s)

	// series2 has Timestamp=101, series3 has Timestamp=1001, series4 has Timestamp=2001.
	// A time range [100, 2000) should match series2 and series3 only.
	tr := timestamp.NewTimeRange(time.Unix(0, 100), time.Unix(0, 2000), true, false)

	t.Run("nil secondary query", func(t *testing.T) {
		query, buildErr := s.BuildQuery(nil, nil, &tr)
		require.NoError(t, buildErr)
		require.NotNil(t, query)

		got, searchErr := s.Search(context.Background(), nil, query, 0)
		require.NoError(t, searchErr)

		sort.Slice(got, func(i, j int) bool {
			return string(got[i].Key.EntityValues) < string(got[j].Key.EntityValues)
		})
		want := []index.SeriesDocument{
			{Key: index.Series{EntityValues: []byte("test2")}, Timestamp: int64(101)},
			{Key: index.Series{EntityValues: []byte("test3")}, Timestamp: int64(1001)},
		}
		assert.Equal(t, want, got)
	})

	t.Run("with secondary query", func(t *testing.T) {
		// Build a secondary query matching entity "test4" (timestamp=2001).
		// The time range [100, 2000) excludes its timestamp, so nothing should match.
		durQuery, buildErr := s.BuildQuery([]index.SeriesMatcher{{
			Type:  index.SeriesMatcherTypeExact,
			Match: []byte("test4"),
		}}, nil, nil)
		require.NoError(t, buildErr)

		// Now query with no series matchers, that secondary query, and the narrow time range.
		combined, buildErr := s.BuildQuery(nil, durQuery, &tr)
		require.NoError(t, buildErr)
		require.NotNil(t, combined)

		got, searchErr := s.Search(context.Background(), nil, combined, 0)
		require.NoError(t, searchErr)
		assert.Empty(t, got, "time range [100,2000) must exclude test4 (timestamp=2001)")
	})
}

func updateData(tester *require.Assertions, s index.SeriesStore) {
	b1, b2 := generateDocs()
	tester.NoError(s.UpdateSeriesBatch(b1))
	tester.NoError(s.UpdateSeriesBatch(b2))
}

func TestStore_StoredFields(t *testing.T) {
	tester := require.New(t)
	path, fn := setUp(tester)
	s, err := NewStore(StoreOpts{
		Path:   path,
		Logger: logger.GetLogger("test"),
	})
	tester.NoError(err)
	defer func() {
		tester.NoError(s.Close())
		fn()
	}()
	insertData(tester, s)
	ctx := context.TODO()

	// No projection: every stored field of test2 is returned, internal
	// bookkeeping fields (_id, _timestamp, ...) excluded.
	all, err := s.StoredFields(ctx, []byte("test2"))
	tester.NoError(err)
	tester.Equal([][]byte{convert.Int64ToBytes(100)}, all[fieldKeyDuration.Marshal()])
	tester.Equal([][]byte{[]byte("svc2")}, all[fieldKeyServiceName.Marshal()])
	tester.Equal([][]byte{convert.Int64ToBytes(100)}, all[fieldKeyStartTime.Marshal()])
	tester.NotContains(all, docIDField)
	tester.NotContains(all, timestampField)

	// Projection: only the requested field is returned.
	only, err := s.StoredFields(ctx, []byte("test2"), fieldKeyDuration)
	tester.NoError(err)
	tester.Len(only, 1)
	tester.Equal([][]byte{convert.Int64ToBytes(100)}, only[fieldKeyDuration.Marshal()])

	// A two-field projection returns exactly those two.
	two, err := s.StoredFields(ctx, []byte("test2"), fieldKeyDuration, fieldKeyServiceName)
	tester.NoError(err)
	tester.Len(two, 2)
	tester.Contains(two, fieldKeyDuration.Marshal())
	tester.Contains(two, fieldKeyServiceName.Marshal())

	// A missing document yields nil.
	none, err := s.StoredFields(ctx, []byte("does-not-exist"))
	tester.NoError(err)
	tester.Nil(none)

	// An array indexed tag stores several values under one field name; they all
	// come back (the result is [][]byte per field, not a single value).
	arrKey := index.FieldKey{IndexRuleID: 4099}
	tester.NoError(s.InsertSeriesBatch(index.Batch{Documents: []index.Document{{
		EntityValues: []byte("arr1"),
		Fields: []index.Field{
			field(arrKey, []byte("a"), true),
			field(arrKey, []byte("b"), true),
			field(fieldKeyServiceName, []byte("svcArr"), true),
		},
	}}}))
	arr, err := s.StoredFields(ctx, []byte("arr1"))
	tester.NoError(err)
	tester.ElementsMatch([][]byte{[]byte("a"), []byte("b")}, arr[arrKey.Marshal()])
	tester.Equal([][]byte{[]byte("svcArr")}, arr[fieldKeyServiceName.Marshal()])

	// Internal bookkeeping fields are excluded even when explicitly projected.
	internalProj, err := s.StoredFields(ctx, []byte("test2"), index.FieldKey{TagName: docIDField}, fieldKeyDuration)
	tester.NoError(err)
	tester.NotContains(internalProj, docIDField)
	tester.Equal([][]byte{convert.Int64ToBytes(100)}, internalProj[fieldKeyDuration.Marshal()])
}
