// Copyright 2022 PingCAP, Inc.
//
// 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 performance

import (
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/workload"
	. "github.com/pingcap/endless/testcase/geopartition/utils"
	"github.com/pingcap/test-infra/sdk/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func localStaleReadTest(ctx *SuiteContext) {
	It("Performance local stale read test", func() {
		defer func() {
			MustExec(ctx.DB, "DROP TABLE IF EXISTS t1")
			MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p1")
		}()
		MustExec(ctx.DB, "DROP TABLE IF EXISTS t1")
		MustExec(ctx.DB, "DROP PLACEMENT POLICY IF EXISTS p1")
		MustExec(ctx.DB, "CREATE PLACEMENT POLICY p1 PRIMARY_REGION='us-east-1' REGIONS='us-east-1,us-east-2,us-west-1'")
		MustExec(ctx.DB, "CREATE TABLE t1 (id int primary key, v int) PLACEMENT POLICY p1")

		MustExec(ctx.DB, "INSERT INTO t1 values(1, 10)")
		time.Sleep(time.Millisecond * 100)
		MustExec(ctx.DB, "SET @a=now(6)")
		time.Sleep(time.Millisecond * 100)
		MustExec(ctx.DB, "UPDATE t1 SET v=11 WHERE id=1")

		// enable chaos
		tc := ctx.TC
		chaosClient := tc.ChaosClient()
		selector := Try(tc.ChaosSelect(resource.TiDBClusterSelectorOpts{
			TiDBClusterTiKVSelectorSpec: &resource.TiDBClusterTiKVSelectorSpec{
				StoreLabelSelector: &metav1.LabelSelector{
					MatchLabels: map[string]string{"region": "us-east-1"},
				},
			},
		})).([]resource.ChaosSelector)

		chaosId := Try(chaosClient.Create(resource.InjectChaosOptions{
			Selector: selector,
			ExperimentSpec: &resource.NetworkLatencySpec{
				Latency: "100ms",
			},
		})).(resource.ChaosIdentifier)
		defer func() {
			Try(tc.ChaosClient().Delete(chaosId))
		}()

		// Check stale read result
		rows := ReadAll(MustQuery(ctx.DB, "SELECT * FROM t1 AS OF TIMESTAMP @a"))
		Expect(rows).Should(Equal([][]string{{"1", "10"}}))

		success := false
		// Because stale read force use mix (leader-and-follower) replica read mode, so we test it 10 times
		// As long as one time the latency < 100ms, it should be seen as succeed
		for i := 0; i < 10; i++ {
			durationStr := ReadAll(MustQuery(ctx.DB, "TRACE SELECT * FROM t1 AS OF TIMESTAMP @a"))[0][2]
			du := Try(time.ParseDuration(durationStr)).(time.Duration)
			if du < time.Millisecond*100 {
				success = true
				break
			}
		}
		Expect(success).Should(BeTrue(), "local stale read should read from follower at least once")
	})
}