package cases

import (
	"context"
	"errors"
	"io"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/avast/retry-go/v4"
	openapiruntime "github.com/go-openapi/runtime"
	"github.com/tidbcloud/dataflow-service-e2e/internal/log"
	"github.com/tidbcloud/dataflow-service-e2e/internal/model"
)

type BaseCase struct {
	t                    *testing.T
	lgr                  log.SimpleLogger
	ctx                  context.Context
	dependenciesEntities map[string]model.DependencyEntity
}

func (k *BaseCase) FulfillDependencyRequirement(requirementName string, entity model.DependencyEntity) {
	if k.dependenciesEntities == nil {
		k.dependenciesEntities = map[string]model.DependencyEntity{}
	}
	k.dependenciesEntities[requirementName] = entity
}

func (k *BaseCase) TiDBDedicated(name string) *model.TiDBDedicatedEntity {
	return k.dependenciesEntities[name].(*model.TiDBDedicatedEntity)
}

func (k *BaseCase) TiDBPremium(name string) *model.TiDBPremiumEntity {
	return k.dependenciesEntities[name].(*model.TiDBPremiumEntity)
}

func (k *BaseCase) Kafka(name string) *model.KafkaEntity {
	return k.dependenciesEntities[name].(*model.KafkaEntity)
}

func (k *BaseCase) MySQL(name string) *model.MySQLEntity {
	return k.dependenciesEntities[name].(*model.MySQLEntity)
}

func (k *BaseCase) S3(name string) *model.S3Entity {
	return k.dependenciesEntities[name].(*model.S3Entity)
}

func (k *BaseCase) GCS(name string) *model.GcsEntity {
	return k.dependenciesEntities[name].(*model.GcsEntity)
}

func (k *BaseCase) AzureBlobStorage(name string) *model.AzureBlobStorageEntity {
	return k.dependenciesEntities[name].(*model.AzureBlobStorageEntity)
}

func (k *BaseCase) TiDBServerless(name string) *model.TiDBServerlessEntity {
	return k.dependenciesEntities[name].(*model.TiDBServerlessEntity)
}

func (k *BaseCase) Pulsar(name string) *model.PulsarEntity {
	return k.dependenciesEntities[name].(*model.PulsarEntity)
}

func (k *BaseCase) Tags() map[string]string {
	return map[string]string{TagKeyDomain: TagValueDomainCDC}
}

func (k *BaseCase) SetT(t *testing.T) {
	k.t = t
}
func (k *BaseCase) SetLogger(lgr log.SimpleLogger) {
	k.lgr = lgr
}
func (k *BaseCase) SetContext(ctx context.Context) {
	k.ctx = ctx
}

// Context returns the runner context for case execution and interruptible
// polling. Deferred cleanup that must survive a graceful Ctrl-C should use
// CleanupContext instead.
func (k *BaseCase) Context() context.Context {
	if k.ctx == nil {
		return context.Background()
	}
	return k.ctx
}

// CleanupContext returns a context that is independent of the case
// interruption. Case-owned deferred cleanup must use this context after a
// graceful Ctrl-C so it can still delete resources created by the case.
func (k *BaseCase) CleanupContext() context.Context {
	ctx := context.Background()
	if k.lgr != nil {
		return log.WithLogger(ctx, k.lgr)
	}
	return ctx
}

func (k *BaseCase) Logger() log.SimpleLogger {
	return k.lgr
}

func yellow(str string) string {
	return "\033[33m" + str + "\033[0m"
}
func red(str string) string {
	return "\033[31m" + str + "\033[0m"
}
func green(str string) string {
	return "\033[32m" + str + "\033[0m"
}

// interrupted reports whether a graceful interruption (first Ctrl-C) is in
// progress. When true the case body should abort quickly so cleanup can run.
func (k *BaseCase) interrupted() bool {
	ctx := k.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	return ctx.Err() != nil
}

// stopIfInterrupted ends the current subtest after a graceful Ctrl-C. Using
// SkipNow (rather than returning to the case) is important: a case is often a
// chain of helpers, and returning would let subsequent actions run against a
// canceled case context. SkipNow still runs testing.T cleanup handlers.
func (k *BaseCase) stopIfInterrupted(message string, args ...any) bool {
	if !k.interrupted() {
		return false
	}
	k.lgr.Info(message, args...)
	// Case-owned defers run while SkipNow unwinds the subtest. Switch the
	// BaseCase helper context before unwinding so those defers can continue to
	// use RetryDo/Expect/Step for cleanup. Deferred code that captured Context
	// earlier must use CleanupContext explicitly.
	k.ctx = k.CleanupContext()
	k.t.SkipNow()
	return true
}

func (k *BaseCase) Step(name string, stepF func() error) *BaseCase {
	if k.stopIfInterrupted("BEGIN-Step '%s' aborted: graceful interrupt in progress", yellow(name)) {
		return k
	}
	k.lgr.Info("BEGIN-Step '%s'", yellow(name))

	err := stepF()
	if k.stopIfInterrupted("END-Step '%s' aborted by Ctrl-C", yellow(name)) {
		return k
	}
	if err != nil {
		k.lgr.Error("FAIL-Step '%s', error: %s", red(name), err.Error())
		k.t.FailNow()
	}
	k.lgr.Info("END-Step '%s' done", green(name))
	return k
}

// RetryDo used to make api with retry
func (k *BaseCase) RetryDo(action string, doF func() error, attempts uint, delay time.Duration) *BaseCase {
	if k.stopIfInterrupted("BEGIN-RetryDo '%s' aborted: graceful interrupt in progress", yellow(action)) {
		return k
	}
	defaultOpts := []retry.Option{retry.Attempts(attempts), retry.DelayType(retry.FixedDelay), retry.Delay(delay)}
	if k.ctx != nil {
		defaultOpts = append(defaultOpts, retry.Context(k.ctx))
	}
	k.lgr.Info("BEGIN-RetryDo '%s'", yellow(action))
	err := retry.Do(func() error { return stopRetryOnHTTP401(doF()) }, defaultOpts...)
	if k.stopIfInterrupted("END-RetryDo '%s' aborted by Ctrl-C: %v", yellow(action), err) {
		return k
	}
	if err != nil {
		k.lgr.Error("FAIL-RetryDo '%s', error: %s", red(action), err.Error())
		k.t.FailNow()
	}
	k.lgr.Info("END-RetryDo '%s'", green(action))
	return k
}

// Expect will verify if it's reach an expected status, return error if timeout
func (k *BaseCase) Expect(status string, statusF func() error, attempts uint, delay time.Duration) *BaseCase {
	if k.stopIfInterrupted("BEGIN-Expect '%s' aborted: graceful interrupt in progress", yellow(status)) {
		return k
	}
	defaultOpts := []retry.Option{retry.Attempts(attempts), retry.DelayType(retry.FixedDelay), retry.Delay(delay)}
	if k.ctx != nil {
		defaultOpts = append(defaultOpts, retry.Context(k.ctx))
	}
	k.lgr.Info("BEGIN-Expect '%s'", yellow(status))
	err := retry.Do(func() error { return stopRetryOnHTTP401(statusF()) }, defaultOpts...)
	if k.stopIfInterrupted("END-Expect '%s' aborted by Ctrl-C: %v", yellow(status), err) {
		return k
	}
	if err != nil {
		k.lgr.Error("FAIL-Expect '%s', error: %s", red(status), err.Error())
		k.t.FailNow()
	}
	k.lgr.Info("END-Expect '%s'", green(status))
	return k
}

var (
	statusExpected = errors.New("status expected ")
)

func checkStillStatus(statusF func() error) error {
	err := statusF()
	if errors.Is(err, io.ErrUnexpectedEOF) {
		return statusF()
	}
	return err
}

// Still will verify if it's in an expected status for a period of time, return error if it's the status can not be kept
func (k *BaseCase) Still(status string, statusF /*return nil if status as expected*/ func() error, attempts uint, delay time.Duration) *BaseCase {
	if k.stopIfInterrupted("BEGIN-Still '%s' aborted: graceful interrupt in progress", yellow(status)) {
		return k
	}
	start := time.Now()
	timeout := delay * time.Duration(attempts-1)
	k.lgr.Info("BEGIN-Still start to check status '%s' can be kept for %s", yellow(status), timeout.String())

	opts := []retry.Option{
		retry.RetryIf(func(err error) bool {
			return errors.Is(err, statusExpected)
		}),
		retry.LastErrorOnly(true),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(delay),
		retry.Attempts(attempts),
	}
	if k.ctx != nil {
		opts = append(opts, retry.Context(k.ctx))
	}
	err := retry.Do(
		func() error {
			if err := checkStillStatus(statusF); err != nil {
				// stop retry
				return stopRetryOnHTTP401(err)
			}
			return statusExpected
		},
		opts...)

	if k.stopIfInterrupted("END-Still status '%s' aborted by Ctrl-C: %v", yellow(status), err) {
		return k
	}

	if !errors.Is(err, statusExpected) {
		interval := time.Now().Sub(start).String()
		k.lgr.Error("FAIL-Still status '%s' can not be kept for %s, it changed after %s,  error: %s", red(status), timeout, interval, err.Error())
		k.t.FailNow()
	}
	k.lgr.Info("END-Still status '%s' still can be kept after %s", green(status), timeout.String())

	return k
}

// stopRetryOnHTTP401 turns authentication/authorization failures into an
// unrecoverable retry error. Polling a control-plane endpoint cannot repair a
// 401, and repeatedly issuing the same request obscures the actionable auth
// failure.
func stopRetryOnHTTP401(err error) error {
	if err == nil {
		return nil
	}
	var apiErr *openapiruntime.APIError
	if errors.As(err, &apiErr) && apiErr.Code == http.StatusUnauthorized {
		return retry.Unrecoverable(err)
	}
	var response interface{ Code() int }
	if errors.As(err, &response) && response.Code() == http.StatusUnauthorized {
		return retry.Unrecoverable(err)
	}
	message := err.Error()
	if strings.Contains(message, "status 401") || strings.Contains(message, "code: 401") || strings.Contains(message, "code=401") {
		return retry.Unrecoverable(err)
	}
	return err
}