package steps

import (
	"context"
	"fmt"

	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/host"
	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	"github.com/pingcap/test-infra/common/model/resource"
	"go.uber.org/zap"
)

type jointString struct {
	sourceName string
	workerName *string
	finalName  *string
}

func (c *jointString) Execute(ctx context.Context) error {
	*c.finalName = c.sourceName + " " + *c.workerName
	fmt.Printf("joint sourceName %s, workerName: %s, finalName : %s", c.sourceName, *c.workerName, *c.finalName)
	return nil
}

func (c *jointString) String() string {
	return fmt.Sprintf("joint sourceName %s, finalName : %s", c.sourceName, *c.finalName)
}

func NewJointString(sourceName string, workerName *string, finalName *string) *jointString {
	return &jointString{
		sourceName,
		workerName,
		finalName,
	}
}

func (b *Builder) JointString(sourceName string, workerName *string, finalName *string) *Builder {
	b.steps = append(b.steps, NewJointString(sourceName, workerName, finalName))
	return b
}

type generate1GBinlogConsumeDurationSql struct {
	workloadId    string
	benchMarkType string
	version       *string
	hashCode      *string
	gtidStatus    string
	relayStatus   string
	tableSharding string
	duration      *string
	executeSql    *string
}

func (c *generate1GBinlogConsumeDurationSql) Execute(ctx context.Context) error {
	*c.executeSql = "insert into BinlogConsumeDuration value (uuid(), " + c.workloadId + ", " + c.benchMarkType + ", \"" + *c.version + "\", \"" + *c.hashCode + "\", " + c.gtidStatus + ", " + c.relayStatus + ", " + c.tableSharding + ", " + *c.duration + ", NOW());"
	fmt.Printf("version is %s, hashCode is: %s, executeSql is: %s", *c.hashCode, *c.version, *c.executeSql)
	return nil
}

func (c *generate1GBinlogConsumeDurationSql) String() string {
	return fmt.Sprintf("executeSql is: %s", *c.executeSql)
}

func NewGenerate1GBinlogConsumeDurationSql(workloadId, benchMarkType string, version, hashCode *string, gtidStatus, relayStatus, tableSharding string, duration, executeSql *string) *generate1GBinlogConsumeDurationSql {
	return &generate1GBinlogConsumeDurationSql{
		workloadId,
		benchMarkType,
		version,
		hashCode,
		gtidStatus,
		relayStatus,
		tableSharding,
		duration,
		executeSql,
	}
}

func (b *Builder) Generate1GBinlogConsumeDurationSql(workloadId, benchMarkType string, version, hashCode *string, gtidStatus, relayStatus, tableSharding string, duration, executeSql *string) *Builder {
	b.steps = append(b.steps, NewGenerate1GBinlogConsumeDurationSql(workloadId, benchMarkType, version, hashCode, gtidStatus, relayStatus, tableSharding, duration, executeSql))
	return b
}

type generateCDCPerfSql struct {
	source *resource.ResourceHandle
	executeSql *string
}

func (c *generateCDCPerfSql) Execute(ctx context.Context) error {
	testbed := host.TestbedPara
	workload := workload.PerfWorkloadPara

	tidbMetrics := TidbMetrics
	ticdccMetrics := TicdcMetrics
	testbedExecuteSql := fmt.Sprintf("INSERT INTO ticdc_testbed (testbed_name,tikv_version,tikv_commit_hash,ticdc_version,ticdc_commit_hash,sink_uri) VALUES ('%s','%s','%s','%s','%s','%s');|",
		workload.TestbedName, testbed.TiKVVersion, testbed.TiKVHashCommit, testbed.TiCDCVersion, testbed.TiCDCHashCommit, testbed.SinkURI)
	log.Info("Execute test sql",zap.String("", testbedExecuteSql))
	workloadExecuteSql := fmt.Sprintf("INSERT INTO ticdc_workload (testbed_name,workload_name,workload_type,threads,tables_or_warehouse,run_start_time,duration) VALUES ('%s','%s','%s',%d,%d,'%s','%s');|",
		workload.TestbedName, workload.WorkloadName, workload.WorkloadType, workload.Threads, workload.TableOrWarehouseCount, workload.RunStartTime, workload.Duration)
	log.Info("Execute workload sql",zap.String("workload sql", workloadExecuteSql))
	tidbExecuteSql := fmt.Sprintf("INSERT INTO ticdc_tidb_summary_log (testbed_name,qps_total,qps_write,p95_latency) VALUES ('%s',%d,%d,%d);|",
		workload.TestbedName, tidbMetrics.QpsTotal, tidbMetrics.QpsWrite, tidbMetrics.P95Latency)
	log.Info("Execute tidb summary log sql",zap.String("tidb metrics sql", workloadExecuteSql))
	ticdcExecuteSql := fmt.Sprintf("INSERT INTO ticdc_summary_log (testbed_name, checkpoint_ts_lag, sink_write_rows) VALUES ('%s',%d,%d);",
		workload.TestbedName, ticdccMetrics.CheckpointLag, ticdccMetrics.SinkWriteRows)
	log.Info("Execute ticdc summary log sql",zap.String("ticdc metrics sql", ticdcExecuteSql))

	*c.executeSql = testbedExecuteSql + workloadExecuteSql + tidbExecuteSql + ticdcExecuteSql

	return nil
}

func (c *generateCDCPerfSql) String() string {
	return fmt.Sprintf("cdc cluster is: %s \n", c.source)
}

func NewGenerateCDCPerfSql(source *resource.ResourceHandle, executeSql *string) *generateCDCPerfSql {
	return &generateCDCPerfSql{
		source,
		executeSql,
	}
}