package cdc

import (
	"context"
	"fmt"
	"github.com/pingcap/log"
	"runtime"
	"strings"
	"time"

	l4g "github.com/alecthomas/log4go"
	"github.com/avast/retry-go"
	"github.com/dailymotion/allure-go"
	"github.com/segmentio/kafka-go"
	"github.com/tidbcloud/qa-PlatformAPIAuto/common"
	dpCommon "github.com/tidbcloud/qa-PlatformAPIAuto/webScene/common"
)

func (s *CommonTidbToKafkaSceneTest) ExecuteManagerCommandToKafkaStep(commandFunc KafkaManagerCommand) {
	allure.Step(
		allure.Description("execute manage kafka command step"),
		allure.Action(func() {
			conn, err := s.GetKafkaManagerConnection()
			defer conn.Close()
			if err != nil {
				errorMsg := fmt.Errorf("connect to kafka occur error: %s", err)
				l4g.Error(errorMsg)
				allure.FailNow(errorMsg)
			}

			err = commandFunc(conn)
			if err != nil {
				errorMsg := fmt.Errorf("execute manage kafka occur error: %s", err)
				l4g.Error(errorMsg)
				allure.FailNow(errorMsg)
			}
		}),
	)
}

func (s *CommonTidbToKafkaSceneTest) GetKafkaManagerConnection() (*kafka.Conn, error) {
	return common.GenerateKafkaControllerConn(s.DbaasRequestInfo.CommonEnv.CdcKafkaUrl)
}

func (s *CommonTidbToKafkaSceneTest) InitKafka(conn *kafka.Conn) error {
	err := conn.DeleteTopics(s.Topic)
	if err != nil && err == kafka.UnknownTopicOrPartition {
		l4g.Info("not find topic: %s", s.Topic)
		return nil
	}
	return err
}

func (s *CommonTidbToKafkaSceneTest) ReadKafkaDataWithRetryStep(readFunc KafkaReadDataCommand) {
	allure.Step(
		allure.Description("read kafka data step"),
		allure.Action(func() {
			log.Info("read kafka data step ")
			reader, err := s.GetKafkaReader()
			defer reader.Close()
			if err != nil {
				errorMsg := fmt.Errorf("get kafka reader occur error: %s", err)
				l4g.Error(errorMsg)
				allure.FailNow(errorMsg)
			}
			err = retry.Do(
				func() error {
					return readFunc(reader)
				},
				retry.DelayType(retry.FixedDelay),
				retry.Delay(time.Second*1),
				retry.Attempts(60*2),
				retry.OnRetry(func(n uint, err error) {
					l4g.Info("#%d: %s\n", n, err)
				}),
			)

			if err != nil {
				l4g.Error(err)
				allure.FailNow(err)
			}
		}),
	)
}

func (s *CommonTidbToKafkaSceneTest) GetKafkaReader() (reader *kafka.Reader, err error) {
	defer func() {
		if rErr := recover(); rErr != nil {
			var stacktrace string
			for i := 2; ; i++ {
				_, f, l, got := runtime.Caller(i)
				if !got {
					break
				}
				stacktrace += fmt.Sprintf("%s:%d\n", f, l)
			}
			err = fmt.Errorf("get panic when getting kafka readererr: %s, stack: %s", rErr, stacktrace)
		}
	}()

	reader = common.GenerateKafkaReader(
		[]string{s.DbaasRequestInfo.CommonEnv.CdcKafkaUrl},
		common.KafkaTopicOp(s.Topic),
		common.KafkaGroupIDOp(s.GroupID))
	return
}

func (s *CommonTidbToKafkaSceneTest) CheckFullStageKafkaData(reader *kafka.Reader) error {
	ctx, registerCancel := context.WithTimeout(context.Background(), time.Second*5)
	defer registerCancel()
	m, err := reader.ReadMessage(ctx)
	if err != nil {
		return err
	}
	message := string(m.Value)
	l4g.Info("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), message)

	if strings.Contains(message, s.TestSchemaName) && strings.Contains(message, string(dpCommon.TABLE_TEST1)) {
		s.FullStageCheckCount++
	}
	if strings.Contains(message, s.TestSchemaName) && strings.Contains(message, string(dpCommon.TABLE_TEST2)) {
		s.FullStageCheckCount++
	}

	if s.FullStageCheckCount != 3 {
		return fmt.Errorf("expect count is 3, but actual count is %d", s.FullStageCheckCount)
	}
	return nil
}

func (s *CommonTidbToKafkaSceneTest) CheckIncreStageKafkaData(reader *kafka.Reader) error {
	message, err := s.readKafkaMessage(reader)
	if err != nil {
		return err
	}

	if strings.Contains(message, s.TestSchemaName) && strings.Contains(message, string(dpCommon.TABLE_TEST3)) {
		s.IncreStageCheckCount++
	}

	if s.IncreStageCheckCount != 3 {
		return fmt.Errorf("expect count is 3, but actual count is %d", s.IncreStageCheckCount)
	}
	return nil
}

func (s *CommonTidbToKafkaSceneTest) readKafkaMessage(reader *kafka.Reader) (string, error) {
	ctx, registerCancel := context.WithTimeout(context.Background(), time.Second*5)
	defer registerCancel()
	m, err := reader.ReadMessage(ctx)
	if err != nil {
		return "", err
	}
	message := string(m.Value)
	l4g.Info("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), message)
	return message, nil
}