// Copyright 2026 TiKV Project Authors.
//
// Licensed 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 endpoint

import (
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	clientv3 "go.etcd.io/etcd/client/v3"

	"github.com/tikv/pd/pkg/storage/kv"
	"github.com/tikv/pd/pkg/utils/keypath"
)

type loadRangeCall struct {
	key    string
	endKey string
	limit  int
}

type recordingKV struct {
	kv.Base
	loadCalls      []string
	loadRangeCalls []loadRangeCall
}

func (s *recordingKV) Load(key string) (string, error) {
	s.loadCalls = append(s.loadCalls, key)
	return s.Base.Load(key)
}

func (s *recordingKV) LoadRange(key, endKey string, limit int) ([]string, []string, error) {
	s.loadRangeCalls = append(s.loadRangeCalls, loadRangeCall{key: key, endKey: endKey, limit: limit})
	if key == "" {
		return nil, nil, errors.New("unexpected full storage root scan")
	}
	return s.Base.LoadRange(key, endKey, limit)
}

func TestLoadTimestampDefaultGroupAvoidsRootScan(t *testing.T) {
	re := require.New(t)
	base := &recordingKV{Base: kv.NewMemoryKV()}
	storage := NewStorageEndpoint(base, nil)
	globalTS := time.Date(2026, 7, 14, 6, 0, 0, 0, time.UTC)
	localTS1 := globalTS.Add(time.Second)
	localTS2 := globalTS.Add(2 * time.Second)

	re.NoError(storage.SaveTimestamp(keypath.TimestampKey, globalTS))
	re.NoError(storage.SaveTimestamp("lta/dc-1/timestamp", localTS1))
	re.NoError(storage.SaveTimestamp("lta/dc-2/timestamp", localTS2))
	re.NoError(base.Save("keyspaces/meta/00000001", "unrelated"))
	re.NoError(base.Save("lta/dc-3/metadata", "unrelated"))

	ts, err := storage.LoadTimestamp("")
	re.NoError(err)
	re.Equal(localTS2.UnixNano(), ts.UnixNano())
	re.Equal([]string{keypath.TimestampKey}, base.loadCalls)
	re.Equal([]loadRangeCall{{
		key:    "lta/",
		endKey: clientv3.GetPrefixRangeEnd("lta/"),
		limit:  0,
	}}, base.loadRangeCalls)
}

func TestLoadTimestampDefaultGroupReturnsGlobalMaximum(t *testing.T) {
	re := require.New(t)
	base := &recordingKV{Base: kv.NewMemoryKV()}
	storage := NewStorageEndpoint(base, nil)
	localTS := time.Date(2026, 7, 14, 6, 0, 0, 0, time.UTC)
	globalTS := localTS.Add(time.Second)

	re.NoError(storage.SaveTimestamp("lta/dc-1/timestamp", localTS))
	re.NoError(storage.SaveTimestamp(keypath.TimestampKey, globalTS))

	ts, err := storage.LoadTimestamp("")
	re.NoError(err)
	re.Equal(globalTS.UnixNano(), ts.UnixNano())
}