package action

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"github.com/avast/retry-go/v4"
	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/logid"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
)

var Tracer = otel.Tracer("dataflow-workflow")

var (
	DagStillRunningError = bizerr.New(bizerr.ErrCodeUnknown, "dag is running")
	StopRetry            = bizerr.New(bizerr.ErrCodeUnknown, "stop retry")
)

type RetryFun func(ctx context.Context, logger *zap.Logger, run func() error) error

type RunAction interface {
	Params() interface{}
	Run(ctx context.Context, logger *zap.Logger, params interface{}) error
}

type RunAfterAction interface {
	RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error
}

type BaseActionOpt func(config *BaseActionConfig)

type BaseActionConfig struct {
	runRetryFun      RetryFun
	runAfterRetryFun RetryFun
}

type Wrapper struct {
	name     string
	run      RunAction
	runAfter RunAfterAction
	config   *BaseActionConfig
}

func MustNewActionWrapper(name string, action interface{}, opts ...BaseActionOpt) *Wrapper {
	runAction, ok := action.(RunAction)
	if !ok {
		panic("action must implement RunAction interface")
	}

	wrapper := &Wrapper{
		name: name,
		run:  runAction,
		config: &BaseActionConfig{
			runRetryFun:      InvokeRunActionWithExponentialRetry,
			runAfterRetryFun: InvokeRunActionWithInfinitelyRetry,
		},
	}

	if runAfterAction, ok := action.(RunAfterAction); ok {
		wrapper.runAfter = runAfterAction
	}
	for _, opt := range opts {
		opt(wrapper.config)
	}
	return wrapper
}

func (m *Wrapper) Name() string {
	return m.name
}

func (m *Wrapper) ParameterNew() interface{} {
	return m.run.Params()
}

func (m *Wrapper) Run(eCtx run.ExecuteContext, params interface{}) error {
	ctx := CtxWithDagInsID(eCtx.Context())
	ctx, span := Tracer.Start(ctx, GenActionSpanName(m.Name(), "Run"))
	defer span.End()

	logger := GetLoggerV2(ctx).Named(m.Name())
	logger.Info("action run start")

	err := m.config.runRetryFun(ctx, logger, func() error {
		return m.run.Run(ctx, logger, params)
	})
	if err != nil {
		logger.Error("action run error", zap.Error(err))
		return err
	}
	logger.Info("action run finished")
	return nil
}

func (m *Wrapper) RunAfter(eCtx run.ExecuteContext, params interface{}) error {
	if m.runAfter == nil {
		return nil
	}

	ctx := CtxWithDagInsID(eCtx.Context())
	ctx, span := Tracer.Start(ctx, GenActionSpanName(m.Name(), "RunAfter"))
	defer span.End()

	logger := GetLoggerV2(ctx).Named(m.Name())
	logger.Info("action runAfter start")

	err := m.config.runAfterRetryFun(ctx, logger, func() error {
		return m.runAfter.RunAfter(ctx, logger, params)
	})
	if err != nil {
		logger.Error("action runAfter error", zap.Error(err))
		return err
	}
	logger.Info("action runAfter finished")
	return nil
}

// WrapperV2 is similar to Wrapper, but it uses the same interface as run.Action and run.AfterAction. This change
// makes it more flexible to wrape a Action many times.
type WrapperV2 struct {
	run      run.Action
	runAfter run.AfterAction
	config   *BaseActionConfig
}

var _ run.Action = &WrapperV2{}

func MustNewActionWrapperV2(action run.Action, opts ...BaseActionOpt) *WrapperV2 {
	wrapper := &WrapperV2{
		run: action,
		config: &BaseActionConfig{
			runRetryFun:      InvokeRunActionWithExponentialRetry,
			runAfterRetryFun: InvokeRunActionWithInfinitelyRetry,
		},
	}

	if runAfterAction, ok := action.(run.AfterAction); ok {
		wrapper.runAfter = runAfterAction
	}
	for _, opt := range opts {
		opt(wrapper.config)
	}
	return wrapper
}

func (m *WrapperV2) Name() string {
	return m.run.Name()
}

func (m *WrapperV2) Run(eCtx run.ExecuteContext, params interface{}) error {
	ctx := CtxWithDagInsID(eCtx.Context())
	ctx, span := Tracer.Start(ctx, GenActionSpanName(m.Name(), "Run"))
	defer span.End()

	logger := GetLoggerV2(ctx).Named(m.Name())
	logger.Info("action run start")

	err := m.config.runRetryFun(ctx, logger, func() error {
		return m.run.Run(eCtx, params)
	})
	if err != nil {
		logger.Error("action run error", zap.Error(err))
		return err
	}
	logger.Info("action run finished")
	return nil
}

func (m *WrapperV2) RunAfter(eCtx run.ExecuteContext, params interface{}) error {
	if m.runAfter == nil {
		return nil
	}

	ctx := CtxWithDagInsID(eCtx.Context())
	ctx, span := Tracer.Start(ctx, GenActionSpanName(m.Name(), "RunAfter"))
	defer span.End()

	logger := GetLoggerV2(ctx).Named(m.Name())
	logger.Info("action runAfter start")

	err := m.config.runAfterRetryFun(ctx, logger, func() error {
		return m.runAfter.RunAfter(eCtx, params)
	})
	if err != nil {
		logger.Error("action runAfter error", zap.Error(err))
		return err
	}
	logger.Info("action runAfter finished")
	return nil
}

func SetRunRetryFun(retryFun RetryFun) BaseActionOpt {
	return func(config *BaseActionConfig) {
		config.runRetryFun = retryFun
	}
}

func SetRunAfterRetryFun(retryFun RetryFun) BaseActionOpt {
	return func(config *BaseActionConfig) {
		config.runAfterRetryFun = retryFun
	}
}

func GetLogger(ctx run.ExecuteContext) *zap.Logger {
	taskInstance := GetTaskIns(ctx.Context())
	ctx.WithValue(logid.CtxKey, taskInstance.DagInsID)
	return log.GetBaseLogger(ctx.Context())
}

func CtxWithDagInsID(ctx context.Context) context.Context {
	taskInstance := GetTaskIns(ctx)
	return context.WithValue(ctx, logid.CtxKey, taskInstance.DagInsID)
}

func GetLoggerV2(ctx context.Context) *zap.Logger {
	return log.GetBaseLogger(ctx)
}

func GetTaskIns(ctx context.Context) *entity.TaskInstance {
	taskIns, _ := entity.CtxRunningTaskIns(ctx)
	return taskIns
}

func GetUint64FromShareData(shareData run.ShareDataOperator, key string) (uint64, error) {
	str, ok := shareData.Get(key)
	if !ok {
		return 0, fmt.Errorf("can not find %s from share data", key)
	}
	number, err := strconv.ParseUint(str, 10, 64)
	if err != nil {
		return 0, err
	}
	return number, nil
}

func GetUint64FromRunVar(ctx run.ExecuteContext, key string) (uint64, error) {
	str, ok := ctx.GetVar(key)
	if !ok {
		return 0, fmt.Errorf("can not find %s from run var", key)
	}
	number, err := strconv.ParseUint(str, 10, 64)
	if err != nil {
		return 0, err
	}
	return number, nil
}

func GetSliceUint64FromShareData(shareData run.ShareDataOperator, key string) ([]uint64, error) {
	value, ok := shareData.Get(key)
	if !ok {
		return nil, fmt.Errorf("can not find %s", key)
	}
	var uint64s []uint64
	err := json.Unmarshal([]byte(value), &uint64s)
	if err != nil {
		return nil, err
	}
	return uint64s, nil
}

func SetSliceUint64IntoShareData(shareData run.ShareDataOperator, key string, value []uint64) error {
	bytes, err := json.Marshal(value)
	if err != nil {
		return err
	}
	shareData.Set(key, string(bytes))
	return nil
}

func GetSliceStringFromShareData(shareData run.ShareDataOperator, key string) ([]string, error) {
	value, ok := shareData.Get(key)
	if !ok {
		return nil, fmt.Errorf("can not find %s", key)
	}
	var strings []string
	err := json.Unmarshal([]byte(value), &strings)
	if err != nil {
		return nil, err
	}
	return strings, nil
}

func SetSliceStringIntoShareData(shareData run.ShareDataOperator, key string, value []string) error {
	bytes, err := json.Marshal(value)
	if err != nil {
		return err
	}
	shareData.Set(key, string(bytes))
	return nil
}

//nolint:gomnd
func InvokeRunActionWithRetry(ctx context.Context, logger *zap.Logger, run func() error) error {
	return retry.Do(func() error {
		return run()
	},
		retry.Context(ctx),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Second*1),
		retry.Attempts(3),
		retry.OnRetry(func(n uint, err error) {
			logger.Warn("action run error", zap.Uint("retry_times", n), zap.Error(err))
		}),
		retry.LastErrorOnly(true),
		retry.RetryIf(func(err error) bool {
			return err != StopRetry
		}),
	)
}

// InvokeRunActionWithExponentialRetry use exponential backoff to retry about 35s
//
//nolint:gomnd
func InvokeRunActionWithExponentialRetry(ctx context.Context, logger *zap.Logger, run func() error) error {
	return retry.Do(func() error {
		return run()
	},
		retry.Context(ctx),
		retry.DelayType(retry.BackOffDelay),
		retry.Delay(time.Millisecond*500),
		retry.MaxDelay(time.Second*10),
		retry.Attempts(8),
		retry.OnRetry(func(n uint, err error) {
			logger.Warn("action run error", zap.Uint("retry_times", n), zap.Error(err))
		}),
		retry.LastErrorOnly(true),
		retry.RetryIf(func(err error) bool {
			return err != StopRetry
		}),
	)
}

//nolint:gomnd
func InvokeRunActionWithInfinitelyRetry(ctx context.Context, logger *zap.Logger, run func() error) error {
	return retry.Do(func() error {
		return run()
	},
		retry.Context(ctx),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Second*15),
		retry.Attempts(0),
		retry.LastErrorOnly(true),
		retry.OnRetry(func(n uint, err error) {
			logger.Warn("action run error", zap.Uint("retry_times", n), zap.Error(err))
		}),
		retry.RetryIf(func(err error) bool {
			return err != StopRetry
		}),
	)
}

// InvokeRunActionWithInfinitelyExponentialRetry use exponential backoff to retry infinitely
//
// the retry interval is 500ms, 1s, 2s, 4s, 8s, 16s, 16s, 16s, ...
//
//nolint:gomnd
func InvokeRunActionWithInfinitelyExponentialRetry(ctx context.Context, logger *zap.Logger, run func() error) error {
	return retry.Do(func() error {
		return run()
	},
		retry.Context(ctx),
		retry.DelayType(retry.BackOffDelay),
		retry.Delay(time.Millisecond*500),
		retry.MaxDelay(time.Second*16),
		retry.Attempts(0),
		retry.OnRetry(func(n uint, err error) {
			logger.Warn("action run error", zap.Uint("retry_times", n), zap.Error(err))
		}),
		retry.LastErrorOnly(true),
		retry.RetryIf(func(err error) bool {
			return err != StopRetry
		}),
	)
}

func GenActionSpanName(actionName, actionFunName string) string {
	return fmt.Sprintf("dataflow-workflow.%s.%s", actionName, actionFunName)
}