grafana/pkg/storage/unified/search/bleve_test.go

package search

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/blevesearch/bleve/v2"
	index "github.com/blevesearch/bleve_index_api"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/testutil"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	bolterrors "go.etcd.io/bbolt/errors"
	"go.uber.org/atomic"
	"go.uber.org/goleak"

	authlib "github.com/grafana/authlib/types"
	"github.com/grafana/grafana/pkg/apimachinery/identity"
	"github.com/grafana/grafana/pkg/apimachinery/utils"
	"github.com/grafana/grafana/pkg/infra/log"
	authzextv1 "github.com/grafana/grafana/pkg/services/authz/proto/v1"
	"github.com/grafana/grafana/pkg/services/store/kind/dashboard"
	"github.com/grafana/grafana/pkg/services/user"
	"github.com/grafana/grafana/pkg/storage/unified/resource"
	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
	"github.com/grafana/grafana/pkg/storage/unified/search/builders"
)

// This verifies that we close all indexes properly and shutdown all background goroutines from our tests.
// (Except for goroutines running specific functions. If possible we should fix this, esp. our own updateIndexSizeMetric.)
func TestMain(m *testing.M) {
	goleak.VerifyTestMain(m,
		goleak.IgnoreTopFunction("github.com/open-feature/go-sdk/openfeature.(*eventExecutor).startEventListener.func1.1"),
		goleak.IgnoreTopFunction("github.com/blevesearch/bleve_index_api.AnalysisWorker"), // These don't stop when index is closed.
	)
}

func TestBleveBackend(t *testing.T) {
	tmpdir, err := os.MkdirTemp("", "grafana-bleve-test")
	require.NoError(t, err)

	backend, err := NewBleveBackend(BleveOptions{
		Root:          tmpdir,
		FileThreshold: 5, // with more than 5 items we create a file on disk
		ScoringModel:  index.BM25Scoring,
	}, nil)
	require.NoError(t, err)
	t.Cleanup(backend.Stop)

	testBleveBackend(t, backend)
}

func testBleveBackend(t *testing.T, backend *bleveBackend) {
	dashboardskey := &resourcepb.ResourceKey{
		Namespace: "default",
		Group:     "dashboard.grafana.app",
		Resource:  "dashboards",
	}
	folderKey := &resourcepb.ResourceKey{
		Namespace: dashboardskey.Namespace,
		Group:     "folder.grafana.app",
		Resource:  "folders",
	}

	rv := int64(10)
	ctx := identity.WithRequester(context.Background(), &user.SignedInUser{Namespace: "ns"})
	var dashboardsIndex resource.ResourceIndex
	var foldersIndex resource.ResourceIndex

	t.Run("build dashboards", func(t *testing.T) {
		key := dashboardskey
		info, err := builders.DashboardBuilder(func(ctx context.Context, namespace string, blob resource.BlobSupport) (resource.DocumentBuilder, error) {
			return &builders.DashboardDocumentBuilder{
				Namespace:        namespace,
				Blob:             blob,
				Stats:            make(map[string]map[string]int64), // empty stats
				DatasourceLookup: dashboard.CreateDatasourceLookup([]*dashboard.DatasourceQueryResult{{}}),
			}, nil
		})
		require.NoError(t, err)

		index, err := backend.BuildIndex(ctx, resource.NamespacedResource{
			Namespace: key.Namespace,
			Group:     key.Group,
			Resource:  key.Resource,
		}, 2, info.Fields, "test", func(index resource.ResourceIndex) (int64, error) {
			err := index.BulkIndex(&resource.BulkIndexRequest{
				Items: []*resource.BulkIndexItem{
					{
						Action: resource.ActionIndex,
						Doc: &resource.IndexableDocument{
							RV:   1,
							Name: "aaa",
							Key: &resourcepb.ResourceKey{
								Name:      "aaa",
								Namespace: "ns",
								Group:     "dashboard.grafana.app",
								Resource:  "dashboards",
							},
							Title:  "aaa (dash)",
							Folder: "xxx",
							Fields: map[string]any{
								builders.DASHBOARD_PANEL_TYPES:       []string{"timeseries", "table"},
								builders.DASHBOARD_ERRORS_TODAY:      25,
								builders.DASHBOARD_VIEWS_LAST_1_DAYS: 50,
							},
							Labels: map[string]string{
								utils.LabelKeyDeprecatedInternalID: "10", // nolint:staticcheck
							},
							Tags: []string{"aa", "bb"},
							Manager: &utils.ManagerProperties{
								Kind:     utils.ManagerKindRepo,
								Identity: "repo-1",
							},
							Source: &utils.SourceProperties{
								Path:            "path/to/aaa.json",
								Checksum:        "xyz",
								TimestampMillis: 1609462800000, // 2021
							},
						},
					},
					{
						Action: resource.ActionIndex,
						Doc: &resource.IndexableDocument{
							RV:   2,
							Name: "bbb",
							Key: &resourcepb.ResourceKey{
								Name:      "bbb",
								Namespace: "ns",
								Group:     "dashboard.grafana.app",
								Resource:  "dashboards",
							},
							Title:  "bbb (dash)",
							Folder: "xxx",
							Fields: map[string]any{
								builders.DASHBOARD_PANEL_TYPES:       []string{"timeseries"},
								builders.DASHBOARD_ERRORS_TODAY:      40,
								builders.DASHBOARD_VIEWS_LAST_1_DAYS: 100,
							},
							Tags: []string{"aa"},
							Labels: map[string]string{
								"region":                           "east",
								utils.LabelKeyDeprecatedInternalID: "11", // nolint:staticcheck
							},
							Manager: &utils.ManagerProperties{
								Kind:     utils.ManagerKindRepo,
								Identity: "repo-1",
							},
							Source: &utils.SourceProperties{
								Path:            "path/to/bbb.json",
								Checksum:        "hijk",
								TimestampMillis: 1640998800000, // 2022
							},
						},
					},
					{
						Action: resource.ActionIndex,
						Doc: &resource.IndexableDocument{
							RV: 3,
							Key: &resourcepb.ResourceKey{
								Name:      "ccc",
								Namespace: "ns",
								Group:     "dashboard.grafana.app",
								Resource:  "dashboards",
							},
							Name:   "ccc",
							Title:  "ccc (dash)",
							Folder: "zzz",
							Manager: &utils.ManagerProperties{
								Kind:     utils.ManagerKindRepo,
								Identity: "repo2",
							},
							Source: &utils.SourceProperties{
								Path: "path/in/repo2.yaml",
							},
							Fields: map[string]any{},
							Tags:   []string{"aa"},
							Labels: map[string]string{
								"region": "west",
							},
						},
					},
				},
			})
			if err != nil {
				return 0, err
			}
			return rv, nil
		}, nil, false)
		require.NoError(t, err)
		require.NotNil(t, index)
		dashboardsIndex = index

		rsp, err := index.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": true}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: key,
			},
			Limit: 100000,
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: resource.SEARCH_FIELD_TITLE, Desc: true}, // ccc,bbb,aaa
			},
			Facet: map[string]*resourcepb.ResourceSearchRequest_Facet{
				"tags": {
					Field: "tags",
					Limit: 100,
				},
			},
		}, nil, nil)
		require.NoError(t, err)
		require.Nil(t, rsp.Error)
		require.NotNil(t, rsp.Results)
		require.NotNil(t, rsp.Facet)

		resource.AssertTableSnapshot(t, filepath.Join("testdata", "manual-dashboard.json"), rsp.Results)

		// Get the tags facets
		facet, ok := rsp.Facet["tags"]
		require.True(t, ok)
		disp, err := json.MarshalIndent(facet, "", "  ")
		require.NoError(t, err)
		//fmt.Printf("%s\n", disp)
		require.JSONEq(t, `{
			"field": "tags",
			"total": 4,
			"terms": [
				{
					"term": "aa",
					"count": 3
				},
				{
					"term": "bb",
					"count": 1
				}
			]
		}`, string(disp))

		count, _ := index.DocCount(ctx, "", nil)
		assert.Equal(t, int64(3), count)

		count, _ = index.DocCount(ctx, "zzz", nil)
		assert.Equal(t, int64(1), count)

		rsp, err = index.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": true}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: key,
				Labels: []*resourcepb.Requirement{{
					Key:      utils.LabelKeyDeprecatedInternalID, // nolint:staticcheck
					Operator: "in",
					Values:   []string{"10", "11"},
				}},
			},
			Limit: 100000,
		}, nil, nil)
		require.NoError(t, err)
		require.Equal(t, int64(2), rsp.TotalHits)
		require.Equal(t, []string{"aaa", "bbb"}, []string{
			rsp.Results.Rows[0].Key.Name,
			rsp.Results.Rows[1].Key.Name,
		})

		// can get sprinkles fields and sort by them
		rsp, err = index.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": true}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: key,
			},
			Limit:  100000,
			Fields: []string{builders.DASHBOARD_ERRORS_TODAY, builders.DASHBOARD_VIEWS_LAST_1_DAYS, "fieldThatDoesntExist"},
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "fields." + builders.DASHBOARD_VIEWS_LAST_1_DAYS, Desc: true},
			},
		}, nil, nil)
		require.NoError(t, err)
		require.Equal(t, 2, len(rsp.Results.Columns))
		require.Equal(t, builders.DASHBOARD_ERRORS_TODAY, rsp.Results.Columns[0].Name)
		require.Equal(t, builders.DASHBOARD_VIEWS_LAST_1_DAYS, rsp.Results.Columns[1].Name)
		// sorted descending so should start with highest dashboard_views_last_1_days (100)
		val, err := resource.DecodeCell(rsp.Results.Columns[1], 0, rsp.Results.Rows[0].Cells[1])
		require.NoError(t, err)
		require.Equal(t, int64(100), val)

		// check auth will exclude results we don't have access to
		rsp, err = index.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": false}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: key,
			},
			Limit:  100000,
			Fields: []string{builders.DASHBOARD_ERRORS_TODAY, builders.DASHBOARD_VIEWS_LAST_1_DAYS, "fieldThatDoesntExist"},
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "fields." + builders.DASHBOARD_VIEWS_LAST_1_DAYS, Desc: true},
			},
		}, nil, nil)
		require.NoError(t, err)
		require.Equal(t, 0, len(rsp.Results.Rows))

		// Now look for repositories
		found, err := index.ListManagedObjects(ctx, &resourcepb.ListManagedObjectsRequest{
			Kind: "repo",
			Id:   "repo-1",
		}, nil)
		require.NoError(t, err)
		jj, err := json.MarshalIndent(found, "", "  ")
		require.NoError(t, err)
		fmt.Printf("%s\n", string(jj))
		// NOTE "hash" -> "checksum" requires changing the protobuf
		require.JSONEq(t, `{
			"items": [
				{
					"object": {
						"namespace": "ns",
						"group": "dashboard.grafana.app",
						"resource": "dashboards",
						"name": "aaa"
					},
					"path": "path/to/aaa.json",
					"hash": "xyz",
					"time": 1609462800000,
					"title": "aaa (dash)",
					"folder": "xxx"
				},
				{
					"object": {
						"namespace": "ns",
						"group": "dashboard.grafana.app",
						"resource": "dashboards",
						"name": "bbb"
					},
					"path": "path/to/bbb.json",
					"hash": "hijk",
					"time": 1640998800000,
					"title": "bbb (dash)",
					"folder": "xxx"
				}
			]
		}`, string(jj))

		counts, err := index.CountManagedObjects(ctx, nil)
		require.NoError(t, err)
		jj, err = json.MarshalIndent(counts, "", "  ")
		require.NoError(t, err)
		fmt.Printf("%s\n", string(jj))
		require.JSONEq(t, `[
			{
				"kind": "repo",
				"id": "repo-1",
				"group": "dashboard.grafana.app",
				"resource": "dashboards",
				"count": 2
			},
			{
				"kind": "repo",
				"id": "repo2",
				"group": "dashboard.grafana.app",
				"resource": "dashboards",
				"count": 1
			}
		]`, string(jj))
	})

	t.Run("build folders", func(t *testing.T) {
		key := folderKey
		var fields resource.SearchableDocumentFields

		index, err := backend.BuildIndex(ctx, resource.NamespacedResource{
			Namespace: key.Namespace,
			Group:     key.Group,
			Resource:  key.Resource,
		}, 2, fields, "test", func(index resource.ResourceIndex) (int64, error) {
			err := index.BulkIndex(&resource.BulkIndexRequest{
				Items: []*resource.BulkIndexItem{
					{
						Action: resource.ActionIndex,
						Doc: &resource.IndexableDocument{
							RV: 1,
							Key: &resourcepb.ResourceKey{
								Name:      "zzz",
								Namespace: "ns",
								Group:     "folder.grafana.app",
								Resource:  "folders",
							},
							Title: "zzz (folder)",
							Manager: &utils.ManagerProperties{
								Kind:     utils.ManagerKindRepo,
								Identity: "repo-1",
							},
							Source: &utils.SourceProperties{
								Path:            "path/to/folder.json",
								Checksum:        "xxxx",
								TimestampMillis: 300,
							},
							Labels: map[string]string{
								utils.LabelKeyDeprecatedInternalID: "123",
							},
						},
					},
					{
						Action: resource.ActionIndex,
						Doc: &resource.IndexableDocument{
							RV: 2,
							Key: &resourcepb.ResourceKey{
								Name:      "yyy",
								Namespace: "ns",
								Group:     "folder.grafana.app",
								Resource:  "folders",
							},
							Title: "yyy (folder)",
							Labels: map[string]string{
								"region":                           "west",
								utils.LabelKeyDeprecatedInternalID: "321",
							},
						},
					},
				},
			})
			if err != nil {
				return 0, err
			}
			return rv, nil
		}, nil, false)
		require.NoError(t, err)
		require.NotNil(t, index)
		foldersIndex = index

		rsp, err := index.Search(ctx, NewStubAccessClient(map[string]bool{"folders": true}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: key,
			},
			Limit: 100000,
		}, nil, nil)
		require.NoError(t, err)
		require.Nil(t, rsp.Error)
		require.NotNil(t, rsp.Results)
		require.Nil(t, rsp.Facet)

		resource.AssertTableSnapshot(t, filepath.Join("testdata", "manual-folder.json"), rsp.Results)
	})

	t.Run("simple federation", func(t *testing.T) {
		// The other tests must run first to build the indexes
		require.NotNil(t, dashboardsIndex)
		require.NotNil(t, foldersIndex)

		// Use a federated query to get both results together, sorted by title
		rsp, err := dashboardsIndex.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": true, "folders": true}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: dashboardskey,
			},
			Fields: []string{
				"title", "_id",
			},
			Federated: []*resourcepb.ResourceKey{
				folderKey, // This will join in the
			},
			Limit: 100000,
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "title", Desc: false},
			},
			Facet: map[string]*resourcepb.ResourceSearchRequest_Facet{
				"region": {
					Field: "labels.region",
					Limit: 100,
				},
			},
		}, []resource.ResourceIndex{foldersIndex}, nil) // << note the folder index matches the federation request
		require.NoError(t, err)
		require.Nil(t, rsp.Error)
		require.NotNil(t, rsp.Results)
		require.NotNil(t, rsp.Facet)

		// Sorted across two indexes
		sorted := []string{}
		for _, row := range rsp.Results.Rows {
			sorted = append(sorted, string(row.Cells[0]))
		}
		require.Equal(t, []string{
			"aaa (dash)",
			"bbb (dash)",
			"ccc (dash)",
			"yyy (folder)",
			"zzz (folder)",
		}, sorted)

		resource.AssertTableSnapshot(t, filepath.Join("testdata", "manual-federated.json"), rsp.Results)

		facet, ok := rsp.Facet["region"]
		require.True(t, ok)
		disp, err := json.MarshalIndent(facet, "", "  ")
		require.NoError(t, err)
		// fmt.Printf("%s\n", disp)
		// NOTE, the west values come from *both* dashboards and folders
		require.JSONEq(t, `{
			"field": "labels.region",
			"total": 3,
			"missing": 2,
			"terms": [
				{
					"term": "west",
					"count": 2
				},
				{
					"term": "east",
					"count": 1
				}
			]
		}`, string(disp))

		// now only when we have permissions to see dashboards
		rsp, err = dashboardsIndex.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": true, "folders": false}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: dashboardskey,
			},
			Fields: []string{
				"title", "_id",
			},
			Federated: []*resourcepb.ResourceKey{
				folderKey, // This will join in the
			},
			Limit: 100000,
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "title", Desc: false},
			},
			Facet: map[string]*resourcepb.ResourceSearchRequest_Facet{
				"region": {
					Field: "labels.region",
					Limit: 100,
				},
			},
		}, []resource.ResourceIndex{foldersIndex}, nil) // << note the folder index matches the federation request

		require.NoError(t, err)
		require.Equal(t, 3, len(rsp.Results.Rows))
		require.Equal(t, "dashboards", rsp.Results.Rows[0].Key.Resource)
		require.Equal(t, "dashboards", rsp.Results.Rows[1].Key.Resource)
		require.Equal(t, "dashboards", rsp.Results.Rows[2].Key.Resource)

		// now only when we have permissions to see folders
		rsp, err = dashboardsIndex.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": false, "folders": true}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: dashboardskey,
			},
			Fields: []string{
				"title", "_id",
			},
			Federated: []*resourcepb.ResourceKey{
				folderKey, // This will join in the
			},
			Limit: 100000,
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "title", Desc: false},
			},
			Facet: map[string]*resourcepb.ResourceSearchRequest_Facet{
				"region": {
					Field: "labels.region",
					Limit: 100,
				},
			},
		}, []resource.ResourceIndex{foldersIndex}, nil) // << note the folder index matches the federation request

		require.NoError(t, err)
		require.Equal(t, 2, len(rsp.Results.Rows))
		require.Equal(t, "folders", rsp.Results.Rows[0].Key.Resource)
		require.Equal(t, "folders", rsp.Results.Rows[1].Key.Resource)

		// now when we have permissions to see nothing
		rsp, err = dashboardsIndex.Search(ctx, NewStubAccessClient(map[string]bool{"dashboards": false, "folders": false}), &resourcepb.ResourceSearchRequest{
			Options: &resourcepb.ListOptions{
				Key: dashboardskey,
			},
			Fields: []string{
				"title", "_id",
			},
			Federated: []*resourcepb.ResourceKey{
				folderKey, // This will join in the
			},
			Limit: 100000,
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "title", Desc: false},
			},
			Facet: map[string]*resourcepb.ResourceSearchRequest_Facet{
				"region": {
					Field: "labels.region",
					Limit: 100,
				},
			},
		}, []resource.ResourceIndex{foldersIndex}, nil) // << note the folder index matches the federation request

		require.NoError(t, err)
		require.Equal(t, 0, len(rsp.Results.Rows))
	})
}

func TestGetSortFields(t *testing.T) {
	t.Run("will prepend 'fields.' to sort fields when they are dashboard fields", func(t *testing.T) {
		searchReq := &resourcepb.ResourceSearchRequest{
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "views_total", Desc: false},
			},
		}
		sortFields := getSortFields(searchReq)
		assert.Equal(t, []string{"fields.views_total"}, sortFields)
	})
	t.Run("will prepend sort fields with a '-' when sort is Desc", func(t *testing.T) {
		searchReq := &resourcepb.ResourceSearchRequest{
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "views_total", Desc: true},
			},
		}
		sortFields := getSortFields(searchReq)
		assert.Equal(t, []string{"-fields.views_total"}, sortFields)
	})
	t.Run("will not prepend 'fields.' to common fields", func(t *testing.T) {
		searchReq := &resourcepb.ResourceSearchRequest{
			SortBy: []*resourcepb.ResourceSearchRequest_Sort{
				{Field: "description", Desc: false},
			},
		}
		sortFields := getSortFields(searchReq)
		assert.Equal(t, []string{"description"}, sortFields)
	})
}

var _ authlib.AccessClient = (*StubAccessClient)(nil)

func NewStubAccessClient(permissions map[string]bool) *StubAccessClient {
	return &StubAccessClient{resourceResponses: permissions}
}

type StubAccessClient struct {
	resourceResponses map[string]bool // key is the resource name, and bool if what the checker will return
}

func (nc *StubAccessClient) Check(ctx context.Context, id authlib.AuthInfo, req authlib.CheckRequest, folder string) (authlib.CheckResponse, error) {
	return authlib.CheckResponse{Allowed: nc.resourceResponses[req.Resource]}, nil
}

func (nc *StubAccessClient) Compile(ctx context.Context, id authlib.AuthInfo, req authlib.ListRequest) (authlib.ItemChecker, authlib.Zookie, error) {
	return func(name, folder string) bool {
		return nc.resourceResponses[req.Resource]
	}, authlib.NoopZookie{}, nil
}

func (nc StubAccessClient) Read(ctx context.Context, req *authzextv1.ReadRequest) (*authzextv1.ReadResponse, error) {
	return nil, nil
}

func (nc StubAccessClient) Write(ctx context.Context, req *authzextv1.WriteRequest) error {
	return nil
}

func (nc StubAccessClient) BatchCheck(ctx context.Context, req *authzextv1.BatchCheckRequest) (*authzextv1.BatchCheckResponse, error) {
	return nil, nil
}

func TestSafeInt64ToInt(t *testing.T) {
	tests := []struct {
		name    string
		input   int64
		want    int
		wantErr bool
	}{
		{
			name:  "Valid int64 within int range",
			input: 42,
			want:  42,
		},
		{
			name:    "Overflow int64 value",
			input:   math.MaxInt64,
			want:    0,
			wantErr: true,
		},
		{
			name:    "Underflow int64 value",
			input:   math.MinInt64,
			want:    0,
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := safeInt64ToInt(tt.input)
			if tt.wantErr {
				require.Error(t, err)
				return
			}
			require.Equal(t, tt.want, got)
		})
	}
}

func Test_isPathWithinRoot(t *testing.T) {
	tests := []struct {
		name string
		dir  string
		root string
		want bool
	}{
		{
			name: "valid path",
			dir:  "/path/to/my-file/",
			root: "/path/to/",
			want: true,
		},
		{
			name: "valid path without trailing slash",
			dir:  "/path/to/my-file",
			root: "/path/to",
			want: true,
		},
		{
			name: "path with double slashes",
			dir:  "/path//to//my-file/",
			root: "/path/to/",
			want: true,
		},
		{
			name: "invalid path: ..",
			dir:  "/path/../above/",
			root: "/path/to/",
		},
		{
			name: "invalid path: \\",
			dir:  "\\path/to",
			root: "/path/to/",
		},
		{
			name: "invalid path: not under safe dir",
			dir:  "/path/to.txt",
			root: "/path/to/",
		},
		{
			name: "invalid path: empty paths",
			dir:  "",
			root: "/path/to/",
		},
		{
			name: "invalid path: different path",
			dir:  "/other/path/to/my-file/",
			root: "/Some/other/path",
		},
		{
			name: "invalid path: empty safe path",
			dir:  "/path/to/",
			root: "",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.Equal(t, tt.want, isPathWithinRoot(tt.dir, tt.root))
		})
	}
}

const (
	buildVersion         = "12.3.45-789"
	defaultFileThreshold = 5
	defaultIndexCacheTTL = 1 * time.Minute
)

func setupBleveBackend(t *testing.T, options ...setupOption) (*bleveBackend, prometheus.Gatherer) {
	reg := prometheus.NewRegistry()
	metrics := resource.ProvideIndexMetrics(reg)

	opts := BleveOptions{
		FileThreshold: defaultFileThreshold,
		IndexCacheTTL: defaultIndexCacheTTL,
		Logger:        log.NewNopLogger(),
		BuildVersion:  buildVersion,
		ScoringModel:  index.BM25Scoring,
	}
	for _, opt := range options {
		opt(&opts)
	}
	if opts.Root == "" {
		opts.Root = t.TempDir()
	}

	backend, err := NewBleveBackend(opts, metrics)
	require.NoError(t, err)
	require.NotNil(t, backend)
	t.Cleanup(backend.Stop)
	return backend, reg
}

type setupOption func(options *BleveOptions)

func withIndexCacheTTL(ttl time.Duration) setupOption {
	return func(options *BleveOptions) {
		options.IndexCacheTTL = ttl
	}
}

func withFileThreshold(threshold int) setupOption {
	return func(options *BleveOptions) {
		options.FileThreshold = int64(threshold)
	}
}

func withRootDir(root string) setupOption {
	return func(options *BleveOptions) {
		options.Root = root
	}
}

func withOwnsIndexFn(fn func(key resource.NamespacedResource) (bool, error)) setupOption {
	return func(options *BleveOptions) {
		options.OwnsIndex = fn
	}
}

func withIndexMinUpdateInterval(d time.Duration) setupOption {
	return func(options *BleveOptions) {
		options.IndexMinUpdateInterval = d
	}
}

func TestBuildIndexExpiration(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	type testCase struct {
		inMemory         bool
		owned            bool
		ownedCheckError  error
		expectedEviction bool
	}

	cacheTTL := time.Millisecond

	for name, tc := range map[string]testCase{
		"memory index should expire, if owned": {
			inMemory:         true,
			owned:            true,
			expectedEviction: true,
		},
		"memory index should expire, if not owned": {
			inMemory:         true,
			owned:            false,
			expectedEviction: true,
		},
		"memory index should expire, if ownership check fails": {
			inMemory:         true,
			ownedCheckError:  errors.New("error"),
			expectedEviction: true,
		},
		"file index should NOT expire, if owned": {
			inMemory:         false,
			owned:            true,
			expectedEviction: false,
		},
		"file index should expire, if not owned": {
			inMemory:         false,
			owned:            false,
			expectedEviction: true,
		},
		"file index should NOT expire, if ownership check fails": {
			inMemory:         false,
			ownedCheckError:  errors.New("error"),
			expectedEviction: false,
		},
	} {
		t.Run(name, func(t *testing.T) {
			backend, reg := setupBleveBackend(t, withIndexCacheTTL(cacheTTL), withOwnsIndexFn(func(key resource.NamespacedResource) (bool, error) {
				return tc.owned, tc.ownedCheckError
			}))

			size := int64(1)
			if !tc.inMemory {
				size = 100 // above defaultFileTreshold
			}
			builtIndex, err := backend.BuildIndex(context.Background(), ns, size, nil, "test", indexTestDocs(ns, 1, 100), nil, false)
			require.NoError(t, err)

			// Evict indexes.
			backend.runEvictExpiredOrUnownedIndexes(time.Now().Add(5 * time.Minute))

			if tc.expectedEviction {
				idx := backend.GetIndex(ns)
				require.Nil(t, idx)

				_, err = builtIndex.DocCount(context.Background(), "", nil)
				require.ErrorIs(t, err, bleve.ErrorIndexClosed)

				// Verify that there are no open indexes.
				checkOpenIndexes(t, reg, 0, 0)
			} else {
				idx := backend.GetIndex(ns)
				require.NotNil(t, idx)

				cnt, err := builtIndex.DocCount(context.Background(), "", nil)
				require.NoError(t, err)
				require.Equal(t, int64(1), cnt)

				// Verify that index is still open
				if tc.inMemory {
					checkOpenIndexes(t, reg, 1, 0)
				} else {
					checkOpenIndexes(t, reg, 0, 1)
				}
			}
		})
	}
}

func TestCloseAllIndexes(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}
	ns2 := resource.NamespacedResource{
		Namespace: "test2",
		Group:     "group",
		Resource:  "resource",
	}

	tmpDir := t.TempDir()
	backend1, reg := setupBleveBackend(t, withRootDir(tmpDir))
	_, err := backend1.BuildIndex(context.Background(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
	require.NoError(t, err)
	_, err = backend1.BuildIndex(context.Background(), ns2, 1 /* memory based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
	require.NoError(t, err)

	// Verify two open indexes.
	checkOpenIndexes(t, reg, 1, 1)
	backend1.closeAllIndexes()

	// Verify that there are no open indexes after closeAllIndexes call.
	checkOpenIndexes(t, reg, 0, 0)
}

func TestBuildIndex(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	for _, rebuild := range []bool{false, true} {
		testName := fmt.Sprintf("rebuild=%t", rebuild)

		t.Run(testName, func(t *testing.T) {
			tmpDir := t.TempDir()

			const (
				firstIndexDocsCount  = 10
				secondIndexDocsCount = 1000
			)

			{
				backend, _ := setupBleveBackend(t, withFileThreshold(5), withRootDir(tmpDir))
				_, err := backend.BuildIndex(context.Background(), ns, firstIndexDocsCount, nil, "test", indexTestDocs(ns, firstIndexDocsCount, 100), nil, rebuild)
				require.NoError(t, err)
				backend.Stop()
			}

			// Make sure we pass at least 1 nanosecond (alwaysRebuildDueToAge) to ensure that the index needs to be rebuild.
			time.Sleep(1 * time.Millisecond)

			newBackend, _ := setupBleveBackend(t, withFileThreshold(5), withRootDir(tmpDir))
			idx, err := newBackend.BuildIndex(context.Background(), ns, secondIndexDocsCount, nil, "test", indexTestDocs(ns, secondIndexDocsCount, 100), nil, rebuild)
			require.NoError(t, err)

			cnt, err := idx.DocCount(context.Background(), "", nil)
			require.NoError(t, err)
			if rebuild {
				require.Equal(t, int64(secondIndexDocsCount), cnt, "Index has been not rebuilt")
			} else {
				require.Equal(t, int64(firstIndexDocsCount), cnt, "Index has not been reused")
			}
		})
	}
}

func TestRebuildingIndexClosesPreviousCachedIndex(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	for name, testCase := range map[string]struct {
		firstInMemory  bool
		secondInMemory bool
	}{
		"in-memory, in-memory": {true, true},
		"in-memory, file":      {true, false},
		"file, in-memory":      {false, true},
		"file, file":           {false, false},
	} {
		t.Run(name, func(t *testing.T) {
			backend, reg := setupBleveBackend(t, withIndexCacheTTL(time.Nanosecond))

			firstSize := 100
			if testCase.firstInMemory {
				firstSize = 1
			}
			firstIndex, err := backend.BuildIndex(context.Background(), ns, int64(firstSize), nil, "test", indexTestDocs(ns, firstSize, 100), nil, false)
			require.NoError(t, err)

			if testCase.firstInMemory {
				verifyDirEntriesCount(t, backend.getResourceDir(ns), 0)
			} else {
				verifyDirEntriesCount(t, backend.getResourceDir(ns), 1)
			}

			openInMemoryIndexes := 0

			secondSize := 100
			if testCase.secondInMemory {
				secondSize = 1
				openInMemoryIndexes = 1
			}
			secondIndex, err := backend.BuildIndex(context.Background(), ns, int64(secondSize), nil, "test", indexTestDocs(ns, secondSize, 100), nil, false)
			require.NoError(t, err)

			if testCase.secondInMemory {
				verifyDirEntriesCount(t, backend.getResourceDir(ns), 0)
			} else {
				verifyDirEntriesCount(t, backend.getResourceDir(ns), 1)
			}

			// Verify that first and second index are different, and first one is now closed.
			require.NotEqual(t, firstIndex, secondIndex)

			_, err = firstIndex.DocCount(context.Background(), "", nil)
			require.ErrorIs(t, err, bleve.ErrorIndexClosed)

			cnt, err := secondIndex.DocCount(context.Background(), "", nil)
			require.NoError(t, err)
			require.Equal(t, int64(secondSize), cnt)

			checkOpenIndexes(t, reg, openInMemoryIndexes, 1-openInMemoryIndexes)
		})
	}
}

func checkOpenIndexes(t *testing.T, reg prometheus.Gatherer, memory, file int) {
	require.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(fmt.Sprintf(`
		# HELP index_server_open_indexes Number of open indexes per storage type. An open index corresponds to single resource group.
		# TYPE index_server_open_indexes gauge
		index_server_open_indexes{index_storage="memory"} %d
		index_server_open_indexes{index_storage="file"} %d
	`, memory, file)), "index_server_open_indexes"))
}

func verifyDirEntriesCount(t *testing.T, dir string, count int) {
	ents, err := os.ReadDir(dir)
	if err != nil {
		if os.IsNotExist(err) {
			ents = nil
			// This is fine, if dir doesn't exist.
		} else {
			require.NoError(t, err)
		}
	}
	require.Len(t, ents, count)
}

func indexTestDocs(ns resource.NamespacedResource, docs int, listRV int64) resource.BuildFn {
	return func(index resource.ResourceIndex) (int64, error) {
		var items []*resource.BulkIndexItem
		for i := 0; i < docs; i++ {
			items = append(items, &resource.BulkIndexItem{
				Action: resource.ActionIndex,
				Doc: &resource.IndexableDocument{
					Key: &resourcepb.ResourceKey{
						Namespace: ns.Namespace,
						Group:     ns.Group,
						Resource:  ns.Resource,
						Name:      fmt.Sprintf("doc%d", i),
					},
					Title: fmt.Sprintf("Document %d", i),
				},
			})
		}

		err := index.BulkIndex(&resource.BulkIndexRequest{Items: items})
		return listRV, err
	}
}

func updateTestDocs(ns resource.NamespacedResource, docs int) resource.UpdateFn {
	cnt := 0

	return func(context context.Context, index resource.ResourceIndex, sinceRV int64) (newRV int64, updatedDocs int, _ error) {
		cnt++

		var items []*resource.BulkIndexItem
		for i := 0; i < docs; i++ {
			items = append(items, &resource.BulkIndexItem{
				Action: resource.ActionIndex,
				Doc: &resource.IndexableDocument{
					Key: &resourcepb.ResourceKey{
						Namespace: ns.Namespace,
						Group:     ns.Group,
						Resource:  ns.Resource,
						Name:      fmt.Sprintf("doc%d", i),
					},
					Title: fmt.Sprintf("Document %d (gen_%d)", i, cnt),
				},
			})
		}

		err := index.BulkIndex(&resource.BulkIndexRequest{Items: items})
		// Simulate RV increase
		return sinceRV + int64(docs), docs, err
	}
}

func updateTestDocsReturningMillisTimestamp(ns resource.NamespacedResource, docs int) (resource.UpdateFn, *atomic.Int64) {
	cnt := 0
	updateCalls := atomic.NewInt64(0)

	return func(context context.Context, index resource.ResourceIndex, sinceRV int64) (newRV int64, updatedDocs int, _ error) {
		now := time.Now()
		updateCalls.Inc()

		cnt++

		var items []*resource.BulkIndexItem
		for i := 0; i < docs; i++ {
			items = append(items, &resource.BulkIndexItem{
				Action: resource.ActionIndex,
				Doc: &resource.IndexableDocument{
					Key: &resourcepb.ResourceKey{
						Namespace: ns.Namespace,
						Group:     ns.Group,
						Resource:  ns.Resource,
						Name:      fmt.Sprintf("doc%d", i),
					},
					Title: fmt.Sprintf("Document %d (gen_%d)", i, cnt),
				},
			})
		}

		err := index.BulkIndex(&resource.BulkIndexRequest{Items: items})
		return now.UnixMilli(), docs, err
	}, updateCalls
}

func TestCleanOldIndexes(t *testing.T) {
	dir := t.TempDir()

	b, _ := setupBleveBackend(t, withRootDir(dir))

	t.Run("with skip", func(t *testing.T) {
		require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-1/a"), 0750))
		require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-2/b"), 0750))
		require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-3/c"), 0750))

		b.cleanOldIndexes(dir, "index-2")
		files, err := os.ReadDir(dir)
		require.NoError(t, err)
		require.Len(t, files, 1)
		require.Equal(t, "index-2", files[0].Name())
	})

	t.Run("without skip", func(t *testing.T) {
		require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-1/a"), 0750))
		require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-2/b"), 0750))
		require.NoError(t, os.MkdirAll(filepath.Join(dir, "index-3/c"), 0750))

		b.cleanOldIndexes(dir, "")
		files, err := os.ReadDir(dir)
		require.NoError(t, err)
		require.Len(t, files, 0)
	})
}

func TestBleveIndexWithFailures(t *testing.T) {
	t.Run("in-memory index", func(t *testing.T) {
		testBleveIndexWithFailures(t, false)
	})
	t.Run("file-based index", func(t *testing.T) {
		testBleveIndexWithFailures(t, true)
	})
}

func testBleveIndexWithFailures(t *testing.T, fileBased bool) {
	backend, _ := setupBleveBackend(t)

	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	size := int64(1)
	if fileBased {
		// size=100 is above FileThreshold (5), make it a file-based index.
		size = 100
	}
	_, err := backend.BuildIndex(context.Background(), ns, size, nil, "test", func(index resource.ResourceIndex) (int64, error) {
		return 0, fmt.Errorf("fail")
	}, nil, false)
	require.Error(t, err)

	// Even though previous build of the index failed, new building of the index should work.
	_, err = backend.BuildIndex(context.Background(), ns, size, nil, "test", indexTestDocs(ns, int(size), 100), nil, false)
	require.NoError(t, err)
}

func TestIndexUpdate(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	be, _ := setupBleveBackend(t)
	idx, err := be.BuildIndex(t.Context(), ns, defaultFileThreshold*2 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false)
	require.NoError(t, err)

	resp := searchTitle(t, idx, "gen", 10, ns)
	require.Equal(t, int64(0), resp.TotalHits)

	// Update index.
	_, err = idx.UpdateIndex(context.Background())
	require.NoError(t, err)

	// Verify that index was updated -- number of docs didn't change, but we can search "gen_1" documents now.
	require.Equal(t, 10, docCount(t, idx))
	require.Equal(t, int64(5), searchTitle(t, idx, "gen_1", 10, ns).TotalHits)

	// Update index again.
	_, err = idx.UpdateIndex(context.Background())
	require.NoError(t, err)
	// Verify that index was updated again -- we can search "gen_2" now. "gen_1" documents are gone.
	require.Equal(t, 10, docCount(t, idx))
	require.Equal(t, int64(0), searchTitle(t, idx, "gen_1", 10, ns).TotalHits)
	require.Equal(t, int64(5), searchTitle(t, idx, "gen_2", 10, ns).TotalHits)
}

func TestConcurrentIndexUpdateAndBuildIndex(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	be, _ := setupBleveBackend(t)

	updaterFn := func(context context.Context, index resource.ResourceIndex, sinceRV int64) (newRV int64, updatedDocs int, _ error) {
		var items []*resource.BulkIndexItem
		for i := 0; i < 5; i++ {
			items = append(items, &resource.BulkIndexItem{
				Action: resource.ActionIndex,
				Doc: &resource.IndexableDocument{
					Key: &resourcepb.ResourceKey{
						Namespace: ns.Namespace,
						Group:     ns.Group,
						Resource:  ns.Resource,
						Name:      fmt.Sprintf("doc%d", i),
					},
					Title: fmt.Sprintf("Document %d (gen_%d)", i, 5),
				},
			})
		}

		err := index.BulkIndex(&resource.BulkIndexRequest{Items: items})
		// Simulate RV increase
		return sinceRV + int64(5), 5, err
	}

	idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false)
	require.NoError(t, err)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_, err = idx.UpdateIndex(ctx)
	require.NoError(t, err)

	_, err = be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false)
	require.NoError(t, err)

	_, err = idx.UpdateIndex(ctx)
	require.Contains(t, err.Error(), bleve.ErrorIndexClosed.Error())
}

func TestConcurrentIndexUpdateSearchAndRebuild(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	be, _ := setupBleveBackend(t)

	_, err := be.BuildIndex(t.Context(), ns, 10, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false)
	require.NoError(t, err)

	wg := sync.WaitGroup{}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	rebuilds := atomic.NewInt64(0)
	updates := atomic.NewInt64(0)
	searches := atomic.NewInt64(0)
	const searchConcurrency = 25
	for i := 0; i < searchConcurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			for ctx.Err() == nil {
				select {
				case <-ctx.Done():
					return
				case <-time.After(time.Duration(i) * time.Millisecond): // introduce small jitter
				}

				idx := be.GetIndex(ns)
				_, err = idx.UpdateIndex(ctx)
				if err != nil {
					if errors.Is(err, bleve.ErrorIndexClosed) || errors.Is(err, context.Canceled) {
						continue
					}
					require.NoError(t, err)
				}
				updates.Inc()

				resp, err := idx.Search(ctx, nil, &resourcepb.ResourceSearchRequest{
					Options: &resourcepb.ListOptions{
						Key: &resourcepb.ResourceKey{
							Namespace: ns.Namespace,
							Group:     ns.Group,
							Resource:  ns.Resource,
						},
					},
					Fields: []string{"title"},
					Query:  "Document",
					Limit:  10,
				}, nil, nil)
				if err != nil {
					if errors.Is(err, bleve.ErrorIndexClosed) || errors.Is(err, context.Canceled) {
						continue
					}
					require.NoError(t, err)
				}
				require.Equal(t, int64(10), resp.TotalHits)
				searches.Inc()
			}
		}()
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		for ctx.Err() == nil {
			_, err := be.BuildIndex(t.Context(), ns, 10, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false)
			require.NoError(t, err)
			rebuilds.Inc()
		}
	}()

	time.Sleep(5 * time.Second)
	cancel()
	wg.Wait()

	t.Log("Updates:", updates.Load(), "searches:", searches.Load(), "rebuilds:", rebuilds.Load())
}

// Verify concurrent updates and searches work as expected.
func TestConcurrentIndexUpdateAndSearch(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	be, _ := setupBleveBackend(t)

	idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateTestDocs(ns, 5), false)
	require.NoError(t, err)

	wg := sync.WaitGroup{}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// We count how many goroutines received given updated RV. We expect at least some RVs to be returned to multiple
	// goroutines, if batching works.
	mu := sync.Mutex{}
	updatedRVs := map[int64]int{}

	const searchConcurrency = 25
	for i := 0; i < searchConcurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			prevRV := int64(0)
			for ctx.Err() == nil {
				// We use t.Context() here to avoid getting errors from context cancellation.
				rv, err := idx.UpdateIndex(t.Context())
				require.NoError(t, err)
				require.Greater(t, rv, prevRV) // Each update should return new RV (that's how our update function works)
				require.Equal(t, int64(10), searchTitle(t, idx, "Document", 10, ns).TotalHits)
				prevRV = rv

				mu.Lock()
				updatedRVs[rv]++
				mu.Unlock()
			}
		}()
	}

	time.Sleep(1 * time.Second)
	cancel()
	wg.Wait()

	// Check that some RVs were updated due to requests from multiple goroutines
	var rvUpdatedByMultipleGoroutines int64
	for rv, count := range updatedRVs {
		if count > 1 {
			rvUpdatedByMultipleGoroutines = rv
			break
		}
	}
	require.Greater(t, rvUpdatedByMultipleGoroutines, int64(0))
}

func TestConcurrentIndexUpdateAndSearchWithIndexMinUpdateInterval(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	const minInterval = 100 * time.Millisecond
	be, _ := setupBleveBackend(t, withIndexMinUpdateInterval(minInterval))

	updateFn, updateCalls := updateTestDocsReturningMillisTimestamp(ns, 5)
	idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updateFn, false)
	require.NoError(t, err)

	wg := sync.WaitGroup{}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	attemptedUpdates := atomic.NewInt64(0)

	// Verify that each returned RV (unix timestamp in millis) is either the same as before, or at least minInterval later.
	const searchConcurrency = 10
	for i := 0; i < searchConcurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			var collectedRVs []int64
			for ctx.Err() == nil {
				attemptedUpdates.Inc()

				// We use t.Context() here to avoid getting errors from context cancellation.
				rv, err := idx.UpdateIndex(t.Context())
				require.NoError(t, err)

				if len(collectedRVs) == 0 || collectedRVs[len(collectedRVs)-1] != rv {
					collectedRVs = append(collectedRVs, rv)
				}

				require.Equal(t, int64(10), searchTitle(t, idx, "Document", 10, ns).TotalHits)
			}

			t.Log(collectedRVs)
			for i := 1; i < len(collectedRVs); i++ {
				// We allow next RV to be 0.9*minInterval later, to account for possible clock skew between time measurements.
				// (We get measurements from update function, but check is done on times inside updater)
				require.GreaterOrEqual(t, collectedRVs[i], collectedRVs[i-1]+(9*int64(minInterval/time.Millisecond)/10))
			}
		}()
	}

	// Run updates and searches for this time.
	testTime := 1 * time.Second

	time.Sleep(testTime)
	cancel()
	wg.Wait()

	expectedMaxCalls := int64(testTime / minInterval)
	require.LessOrEqual(t, updateCalls.Load(), expectedMaxCalls+1)
	require.Greater(t, attemptedUpdates.Load(), updateCalls.Load())

	t.Log("Attempted updates:", attemptedUpdates.Load(), "update calls:", updateCalls.Load())
}

func TestIndexUpdateWithErrors(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	be, _ := setupBleveBackend(t)

	updateErr := fmt.Errorf("failed to update index")
	updaterFn := func(context context.Context, index resource.ResourceIndex, sinceRV int64) (newRV int64, updatedDocs int, _ error) {
		time.Sleep(100 * time.Millisecond)
		return 0, 0, updateErr
	}
	idx, err := be.BuildIndex(t.Context(), ns, 10 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), updaterFn, false)
	require.NoError(t, err)

	t.Run("update fail", func(t *testing.T) {
		_, err = idx.UpdateIndex(t.Context())
		require.ErrorIs(t, err, updateErr)
	})

	t.Run("update timeout", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
		defer cancel()

		_, err = idx.UpdateIndex(ctx)
		require.ErrorIs(t, err, context.DeadlineExceeded)
	})

	t.Run("context canceled", func(t *testing.T) {
		// Canceled context
		ctx, cancel := context.WithCancel(context.Background())
		cancel()

		_, err = idx.UpdateIndex(ctx)
		require.ErrorIs(t, err, context.Canceled)
	})
}

func TestIndexBuildInfo(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	be, _ := setupBleveBackend(t, withFileThreshold(100))
	index, err := be.BuildIndex(t.Context(), ns, 10, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
	require.NoError(t, err)

	buildInfo, err := getBuildInfo(index.(*bleveIndex).index)
	require.NoError(t, err)
	require.NotNil(t, buildInfo)
	require.Equal(t, buildVersion, buildInfo.BuildVersion)
	require.InDelta(t, float64(time.Now().Unix()), buildInfo.BuildTime, 30) // allow 30 seconds of drift
}

func TestInvalidBuildVersion(t *testing.T) {
	opts := BleveOptions{
		Root:         t.TempDir(),
		BuildVersion: "invalid",
	}
	_, err := NewBleveBackend(opts, nil)
	require.ErrorContains(t, err, "cannot parse build version")
}

func searchTitle(t *testing.T, idx resource.ResourceIndex, query string, limit int, ns resource.NamespacedResource) *resourcepb.ResourceSearchResponse {
	resp, err := idx.Search(t.Context(), nil, &resourcepb.ResourceSearchRequest{
		Options: &resourcepb.ListOptions{
			Key: &resourcepb.ResourceKey{
				Namespace: ns.Namespace,
				Group:     ns.Group,
				Resource:  ns.Resource,
			},
		},
		Fields: []string{"title"},
		Query:  query,
		Limit:  int64(limit),
	}, nil, nil)
	require.NoError(t, err)
	return resp
}

func docCount(t *testing.T, idx resource.ResourceIndex) int {
	cnt, err := idx.DocCount(context.Background(), "", nil)
	require.NoError(t, err)
	return int(cnt)
}

func TestBuildIndexReturnsErrorWhenIndexLocked(t *testing.T) {
	ns := resource.NamespacedResource{
		Namespace: "test",
		Group:     "group",
		Resource:  "resource",
	}

	tmpDir := t.TempDir()

	// First, create a file-based index with one backend and keep it open
	backend1, reg1 := setupBleveBackend(t, withRootDir(tmpDir))
	index1, err := backend1.BuildIndex(context.Background(), ns, 100 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
	require.NoError(t, err)
	require.NotNil(t, index1)

	// Verify first index is file-based
	bleveIdx1, ok := index1.(*bleveIndex)
	require.True(t, ok)
	require.Equal(t, indexStorageFile, bleveIdx1.indexStorage)
	checkOpenIndexes(t, reg1, 0, 1)

	// Now create a second backend using the same directory
	// This simulates another instance trying to open the same index
	backend2, _ := setupBleveBackend(t, withRootDir(tmpDir))

	// BuildIndex should detect the file is locked and return an error after timeout
	now := time.Now()
	timeout, err := time.ParseDuration(boltTimeout)
	require.NoError(t, err)
	index2, err := backend2.BuildIndex(context.Background(), ns, 100 /* file based */, nil, "test", indexTestDocs(ns, 10, 100), nil, false)
	require.Error(t, err)
	require.ErrorIs(t, err, bolterrors.ErrTimeout)
	require.Nil(t, index2)
	require.GreaterOrEqual(t, time.Since(now).Milliseconds(), timeout.Milliseconds()-500, "BuildIndex should have waited for approximately boltTimeout duration")

	// Clean up: close first backend to release the file lock
	backend1.Stop()
}