package testcases

import (
	"time"

	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	"github.com/pingcap/test-infra/caselib/pkg/steps"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	"github.com/pingcap/test-infra/common/model/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
)

var _ = AddTestCase("binlog_kafka_large_transcation", func() steps.Step {
	var (
		UpStream   = &resource.ResourceHandle{}
		DownStream = &resource.ResourceHandle{}
		WorkLoad   = &resource.ResourceHandle{}
		Arbiter    = &resource.ResourceHandle{}
		DBName     = "binlog_kafka_large_transcation"
		executeSql string
	)

	return steps.
		NewBuilder().
		LoadEnv().
		LoadResourceFromEnv(utils.ResourcePrefix+"upstream", UpStream).
		LoadResourceFromEnv(utils.ResourcePrefix+"downstream", DownStream).
		LoadResourceFromEnv(utils.ResourcePrefix+"workload", WorkLoad).
		LoadResourceFromEnv(utils.ResourcePrefix+"arbiter", Arbiter).
		StartBinlogArbiter(Arbiter).
		CreateDatabase(UpStream, DBName).
		RunBinlogWorkload(UpStream, WorkLoad, Spec.GetWorkloadOption(workload.CommandTypePrepare)).
		RunBinlogWorkload(UpStream, WorkLoad, Spec.GetWorkloadOption(workload.CommandTypeRun)).
		// Execute big transcation
		ExecuteSqlStatement(UpStream, &executeSql).
		Sleep(5*time.Minute).
		CheckBinlogConsistency(UpStream, DownStream, DBName).
		Build()
})