/*
 * Copyright 2021 CloudWeGo Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package retry

import (
	"context"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/bytedance/sonic"

	"github.com/cloudwego/kitex/internal/test"
	"github.com/cloudwego/kitex/pkg/circuitbreak"
	"github.com/cloudwego/kitex/pkg/discovery"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/pkg/remote"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/rpcinfo/remoteinfo"
	"github.com/cloudwego/kitex/pkg/serviceinfo"
	"github.com/cloudwego/kitex/pkg/stats"
)

var (
	remoteTagKey   = "k"
	remoteTagValue = "v"
	remoteTags     = map[string]string{remoteTagKey: remoteTagValue}
	sonici         = sonic.Config{SortMapKeys: true, UseNumber: true}.Froze()
)

// test new retry container
func TestNewRetryContainer(t *testing.T) {
	rc := NewRetryContainerWithCB(nil, nil)
	rc.NotifyPolicyChange(method, Policy{
		Enable:        true,
		BackupPolicy:  NewBackupPolicy(10),
		FailurePolicy: NewFailurePolicy(),
	})
	r := rc.getRetryer(context.Background(), genRPCInfo())
	_, ok := r.(*failureRetryer)
	test.Assert(t, ok)

	rc.NotifyPolicyChange(method, Policy{
		Enable:        false,
		BackupPolicy:  NewBackupPolicy(10),
		FailurePolicy: NewFailurePolicy(),
	})
	_, ok = r.(*failureRetryer)
	test.Assert(t, ok)
	_, allow := r.AllowRetry(context.Background())
	test.Assert(t, !allow)

	rc.NotifyPolicyChange(method, Policy{
		Enable:        true,
		BackupPolicy:  NewBackupPolicy(10),
		FailurePolicy: NewFailurePolicy(),
	})
	r = rc.getRetryer(context.Background(), genRPCInfo())
	_, ok = r.(*failureRetryer)
	test.Assert(t, ok)
	_, allow = r.AllowRetry(context.Background())
	test.Assert(t, allow)

	rc.NotifyPolicyChange(method, Policy{
		Enable:       true,
		Type:         1,
		BackupPolicy: NewBackupPolicy(20),
	})
	r = rc.getRetryer(context.Background(), genRPCInfo())
	_, ok = r.(*backupRetryer)
	test.Assert(t, ok)
	_, allow = r.AllowRetry(context.Background())
	test.Assert(t, allow)

	rc.NotifyPolicyChange(method, Policy{
		Enable:       false,
		Type:         1,
		BackupPolicy: NewBackupPolicy(20),
	})
	r = rc.getRetryer(context.Background(), genRPCInfo())
	_, ok = r.(*backupRetryer)
	test.Assert(t, ok)
	_, allow = r.AllowRetry(context.Background())
	test.Assert(t, !allow)
}

// test invalid policy
func TestRetryInvalidPolicyInit(t *testing.T) {
	t.Run("type is mixedRetry but policy is not", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":1,
"mixed_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}}
}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, p)
		test.Assert(t, strings.Contains(rc.msg, "BackupPolicy is nil or retry type not match"), rc.msg)
	})

	t.Run("type is backupRequest but policy is not", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":1,
"mixed_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}},
"failure_policy": {}
}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, p)
		test.Assert(t, strings.Contains(rc.msg, "BackupPolicy is nil or retry type not match"), rc.msg)
	})

	t.Run("type is mixedRetry but policy is empty", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":2}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, p)
		test.Assert(t, strings.Contains(rc.msg, "MixedPolicy is nil or retry type not match"), rc.msg)
	})

	t.Run("type is failureRetry but policy is not", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":0,
"mixed_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}}}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, p)
		test.Assert(t, strings.Contains(rc.msg, "FailurePolicy is nil or retry type not match"), rc.msg)
	})

	t.Run("failurePolicy is nil", func(t *testing.T) {
		rc := NewRetryContainer()
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
		})
		test.Assert(t, strings.Contains(rc.msg, "FailurePolicy is nil or retry type not match"))
	})

	t.Run("backupPolicy is nil", func(t *testing.T) {
		cbs := newCBSuite(nil)
		rc := NewRetryContainerWithCBStat(cbs.ServiceControl(), cbs.ServicePanel())
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   1,
		})
		test.Assert(t, strings.Contains(rc.msg, "BackupPolicy is nil or retry type not match"))
	})

	t.Run("mixedPolicy is nil", func(t *testing.T) {
		rc := NewRetryContainer()
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   2,
		})
		test.Assert(t, strings.Contains(rc.msg, "MixedPolicy is nil or retry type not match"))
	})

	t.Run("backupPolicy config invalid", func(t *testing.T) {
		rc := NewRetryContainer()
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   1,
			BackupPolicy: &BackupPolicy{
				RetryDelayMS: 0,
			},
		})
		msg := "new retryer[test-Backup] failed, err=newBackupRetryer failed, err=invalid retry delay duration in backupRetryer, at "
		test.Assert(t, rc.msg[:len(msg)] == msg, rc.msg)
	})

	t.Run("backupPolicy config invalid", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   1,
			BackupPolicy: &BackupPolicy{
				RetryDelayMS: 100,
				StopPolicy: StopPolicy{
					CBPolicy: CBPolicy{
						ErrorRate: 0.4,
					},
				},
			},
		})
		msg := "new retryer[test-Backup] at "
		test.Assert(t, rc.msg[:len(msg)] == msg)
	})

	t.Run("failurePolicy config invalid", func(t *testing.T) {
		cbs := newCBSuite(nil)
		rc := NewRetryContainerWithCBStat(cbs.ServiceControl(), cbs.ServicePanel())
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
			FailurePolicy: &FailurePolicy{
				StopPolicy: StopPolicy{
					MaxRetryTimes: 6,
				},
			},
		})
		msg := "new retryer[test-Failure] failed, err=newfailureRetryer failed, err=invalid MaxRetryTimes[6], at "
		test.Assert(t, rc.msg[:len(msg)] == msg, rc.msg)
	})

	t.Run("failurePolicy cBPolicy config invalid", func(t *testing.T) {
		rc := NewRetryContainer()
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
			FailurePolicy: &FailurePolicy{
				StopPolicy: StopPolicy{
					MaxRetryTimes: 5,
					CBPolicy: CBPolicy{
						ErrorRate: 0.4,
					},
				},
			},
		})
		msg := "new retryer[test-Failure] at "
		test.Assert(t, rc.msg[:len(msg)] == msg)
	})

	t.Run("failurePolicy backOffPolicy fixedBackOffType cfg is nil", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
			FailurePolicy: &FailurePolicy{
				StopPolicy: StopPolicy{
					MaxRetryTimes: 5,
				},
				BackOffPolicy: &BackOffPolicy{
					BackOffType: FixedBackOffType,
				},
			},
		})
		msg := "new retryer[test-Failure] at "
		test.Assert(t, rc.msg[:len(msg)] == msg)
	})

	t.Run("failurePolicy backOffPolicy fixedBackOffType cfg invalid", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
			FailurePolicy: &FailurePolicy{
				StopPolicy: StopPolicy{
					MaxRetryTimes: 5,
				},
				BackOffPolicy: &BackOffPolicy{
					BackOffType: FixedBackOffType,
					CfgItems: map[BackOffCfgKey]float64{
						FixMSBackOffCfgKey: 0,
					},
				},
			},
		})
		msg := "new retryer[test-Failure] at "
		test.Assert(t, rc.msg[:len(msg)] == msg)
	})

	t.Run("failurePolicy backOffPolicy randomBackOffType cfg is nil", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
			FailurePolicy: &FailurePolicy{
				StopPolicy: StopPolicy{
					MaxRetryTimes: 5,
				},
				BackOffPolicy: &BackOffPolicy{
					BackOffType: RandomBackOffType,
				},
			},
		})
		msg := "new retryer[test-Failure] at "
		test.Assert(t, rc.msg[:len(msg)] == msg)
	})

	t.Run("failurePolicy backOffPolicy randomBackOffType cfg invalid", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
			FailurePolicy: &FailurePolicy{
				StopPolicy: StopPolicy{
					MaxRetryTimes: 5,
				},
				BackOffPolicy: &BackOffPolicy{
					BackOffType: RandomBackOffType,
					CfgItems: map[BackOffCfgKey]float64{
						MinMSBackOffCfgKey: 20,
						MaxMSBackOffCfgKey: 10,
					},
				},
			},
		})
		msg := "new retryer[test-Failure] at "
		test.Assert(t, rc.msg[:len(msg)] == msg)
	})

	t.Run("failurePolicy backOffPolicy randomBackOffType normal", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		rc.NotifyPolicyChange(method, Policy{
			Enable: true,
			Type:   0,
			FailurePolicy: &FailurePolicy{
				StopPolicy: StopPolicy{
					MaxRetryTimes: 5,
				},
				BackOffPolicy: &BackOffPolicy{
					BackOffType: RandomBackOffType,
					CfgItems: map[BackOffCfgKey]float64{
						MinMSBackOffCfgKey: 10,
						MaxMSBackOffCfgKey: 20,
					},
				},
			},
		})
		msg := "new retryer[test-Failure] at "
		test.Assert(t, rc.msg[:len(msg)] == msg)
	})

	t.Run("test init invalid case", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		err := rc.Init(nil, nil)
		test.Assert(t, err == nil, err)
		err = rc.Init(map[string]Policy{Wildcard: {
			Enable: true,
			Type:   1,
			BackupPolicy: &BackupPolicy{
				RetryDelayMS: 0,
			},
		}}, nil)
		test.Assert(t, err != nil, err)

		rc.DeletePolicy(method)
		r := rc.getRetryer(context.Background(), genRPCInfo())
		test.Assert(t, r == nil)
	})
}

// test container dump
func TestContainer_Dump(t *testing.T) {
	// test backupPolicy dump
	t.Run("backupPolicy dump", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		methodPolicies := map[string]Policy{
			method: {
				Enable:       true,
				Type:         BackupType,
				BackupPolicy: NewBackupPolicy(20),
			},
		}
		err := rc.Init(methodPolicies, nil)
		test.Assert(t, err == nil, err)
		rcDump, ok := rc.Dump().(map[string]interface{})
		test.Assert(t, ok)
		hasCodeCfg := rcDump["has_code_cfg"].(bool)
		test.Assert(t, hasCodeCfg)
		testStr, err := sonici.MarshalToString(rcDump[method])
		msg := `{"backup_request":{"retry_delay_ms":20,"stop_policy":{"max_retry_times":1,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1}},"retry_same_node":false},"enable":true}`
		test.Assert(t, err == nil, err)
		test.Assert(t, testStr == msg)
	})

	// test backupPolicy dump
	t.Run("backupPolicy dump without code_cfg", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		policy := Policy{
			Enable:       true,
			Type:         1,
			BackupPolicy: NewBackupPolicy(20),
		}
		err := rc.Init(nil, nil)
		rc.NotifyPolicyChange(method, policy)
		test.Assert(t, err == nil, err)
		rcDump, ok := rc.Dump().(map[string]interface{})
		test.Assert(t, ok)
		hasCodeCfg := rcDump["has_code_cfg"].(bool)
		test.Assert(t, !hasCodeCfg)
		testStr, err := sonici.MarshalToString(rcDump[method])
		msg := `{"backup_request":{"retry_delay_ms":20,"stop_policy":{"max_retry_times":1,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1}},"retry_same_node":false},"enable":true}`
		test.Assert(t, err == nil, err)
		test.Assert(t, testStr == msg)
	})

	// test failurePolicy dump
	t.Run("failurePolicy dump", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		methodPolicies := map[string]Policy{
			method: {
				Enable:        true,
				Type:          FailureType,
				FailurePolicy: NewFailurePolicy(),
			},
		}
		err := rc.Init(methodPolicies, nil)
		test.Assert(t, err == nil, err)
		rcDump, ok := rc.Dump().(map[string]interface{})
		test.Assert(t, ok)
		hasCodeCfg := rcDump["has_code_cfg"].(bool)
		test.Assert(t, hasCodeCfg)
		testStr, err := sonici.MarshalToString(rcDump[method])
		msg := `{"enable":true,"failure_retry":{"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":""},"specified_result_retry":{"error_retry":false,"old_error_retry":false,"old_resp_retry":false,"resp_retry":false}}`
		test.Assert(t, err == nil, err)
		test.Assert(t, testStr == msg, testStr)
	})

	// test mixedPolicy dump
	t.Run("mixedPolicy dump", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		policy := Policy{
			Enable:      true,
			Type:        MixedType,
			MixedPolicy: NewMixedPolicy(20),
		}
		err := rc.Init(nil, nil)
		test.Assert(t, err == nil, err)
		rc.NotifyPolicyChange(method, policy)
		rcDump, ok := rc.Dump().(map[string]interface{})
		test.Assert(t, ok)
		hasCodeCfg := rcDump["has_code_cfg"].(bool)
		test.Assert(t, !hasCodeCfg)
		testStr, err := sonici.MarshalToString(rcDump[method])
		msg := `{"enable":true,"mixed_retry":{"retry_delay_ms":20,"stop_policy":{"max_retry_times":1,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":""},"specified_result_retry":{"error_retry":false,"old_error_retry":false,"old_resp_retry":false,"resp_retry":false}}`
		test.Assert(t, err == nil, err)
		test.Assert(t, testStr == msg, testStr)
	})

	// test mixedPolicy dump
	t.Run("mixedPolicy dump with customized retry", func(t *testing.T) {
		rc := NewRetryContainerWithCB(nil, nil)
		policy := Policy{
			Enable:      true,
			Type:        MixedType,
			MixedPolicy: NewMixedPolicy(20),
		}
		rr := &ShouldResultRetry{ErrorRetryWithCtx: func(ctx context.Context, err error, ri rpcinfo.RPCInfo) bool {
			if ri.To().Method() == method {
				if te, ok := err.(*remote.TransError); ok && te.TypeID() == 1000 {
					return true
				}
			}
			return false
		}}
		err := rc.Init(nil, rr)
		test.Assert(t, err == nil, err)
		rc.NotifyPolicyChange(method, policy)
		rcDump, ok := rc.Dump().(map[string]interface{})
		test.Assert(t, ok)
		hasCodeCfg := rcDump["has_code_cfg"].(bool)
		test.Assert(t, !hasCodeCfg)
		testStr, err := sonici.MarshalToString(rcDump[method])
		msg := `{"enable":true,"mixed_retry":{"retry_delay_ms":20,"stop_policy":{"max_retry_times":1,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":""},"specified_result_retry":{"error_retry":true,"old_error_retry":false,"old_resp_retry":false,"resp_retry":false}}`
		test.Assert(t, err == nil, err)
		test.Assert(t, testStr == msg, testStr)
	})
}

// test different method use different retry policy
func TestDifferentMethodConfig(t *testing.T) {
	var callTimes int32
	methodRetryer := make(map[string]Retryer)
	var lock sync.Mutex
	rpcCall := func(ctx context.Context, r Retryer, request, response interface{}) (rpcinfo.RPCInfo, error) {
		atomic.AddInt32(&callTimes, 1)
		ri := rpcinfo.GetRPCInfo(ctx)
		if ri.To().Method() == method {
			lock.Lock()
			methodRetryer[ri.To().Method()] = r
			lock.Unlock()
			// specified method mock return error and do error retry
			if atomic.LoadInt32(&callTimes) == 1 {
				return genRPCInfo(), remote.NewTransErrorWithMsg(1000, "mock")
			}
		} else {
			lock.Lock()
			methodRetryer[ri.To().Method()] = r
			lock.Unlock()
			if atomic.LoadInt32(&callTimes) == 1 {
				// other method do backup request
				time.Sleep(20 * time.Millisecond)
				return genRPCInfo(), kerrors.ErrRPCTimeout
			}
		}
		return genRPCInfo(), nil
	}
	rc := NewRetryContainer()
	shouldResultRetry := &ShouldResultRetry{ErrorRetry: func(err error, ri rpcinfo.RPCInfo) bool {
		if ri.To().Method() == method {
			if te, ok := err.(*remote.TransError); ok && te.TypeID() == 1000 {
				return true
			}
		}
		return false
	}}
	err := rc.Init(map[string]Policy{
		method:   BuildFailurePolicy(NewFailurePolicy()),
		Wildcard: BuildBackupRequest(NewBackupPolicy(10)),
	}, shouldResultRetry)
	test.Assert(t, err == nil, err)

	// case1: test method do error retry
	ri := genRPCInfo()
	ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
	_, ok, err := rc.WithRetryIfNeeded(ctx, &Policy{}, rpcCall, ri, nil, nil)
	test.Assert(t, err == nil, err)
	test.Assert(t, !ok)
	lock.Lock()
	_, ok = methodRetryer[method].(*failureRetryer)
	lock.Unlock()
	test.Assert(t, ok)

	// case2: other method do backup request
	method2 := "method2"
	to := remoteinfo.NewRemoteInfo(&rpcinfo.EndpointBasicInfo{Method: method2}, method2).ImmutableView()
	ri = rpcinfo.NewRPCInfo(to, to, rpcinfo.NewInvocation("", method2), rpcinfo.NewRPCConfig(), rpcinfo.NewRPCStats())
	setter := ri.Invocation().(rpcinfo.InvocationSetter)
	setter.SetMethodInfo(serviceinfo.NewMethodInfo(nil, nil, func() interface{} {
		return nil
	}, false))
	ctx = rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
	_, ok, err = rc.WithRetryIfNeeded(ctx, &Policy{}, rpcCall, ri, nil, nil)
	test.Assert(t, err == nil, err)
	test.Assert(t, !ok)
	lock.Lock()
	_, ok = methodRetryer[method2].(*backupRetryer)
	lock.Unlock()
	test.Assert(t, ok)
}

// test policy noRetry call
func TestPolicyNoRetryCall(t *testing.T) {
	ctx := context.Background()
	rc := NewRetryContainer()

	// case 1(default): no retry policy
	// no retry policy, call success
	var callTimes int32
	ri := genRPCInfo()
	ctx = rpcinfo.NewCtxWithRPCInfo(ctx, ri)
	_, ok, err := rc.WithRetryIfNeeded(ctx, &Policy{}, func(ctx context.Context, r Retryer, request, response interface{}) (rpcinfo.RPCInfo, error) {
		atomic.AddInt32(&callTimes, 1)
		return ri, nil
	}, ri, nil, nil)
	test.Assert(t, err == nil, err)
	test.Assert(t, atomic.LoadInt32(&callTimes) == 1)
	test.Assert(t, ok)

	// no retry policy, call rpcTimeout
	atomic.StoreInt32(&callTimes, 0)
	ri = genRPCInfo()
	ctx = rpcinfo.NewCtxWithRPCInfo(ctx, ri)
	_, ok, err = rc.WithRetryIfNeeded(ctx, &Policy{}, func(ctx context.Context, r Retryer, request, response interface{}) (rpcinfo.RPCInfo, error) {
		atomic.AddInt32(&callTimes, 1)
		if atomic.LoadInt32(&callTimes) == 1 {
			return ri, kerrors.ErrRPCTimeout
		}
		return ri, nil
	}, ri, nil, nil)
	test.Assert(t, kerrors.IsTimeoutError(err), err)
	test.Assert(t, atomic.LoadInt32(&callTimes) == 1)
	test.Assert(t, !ok)

	// case 2: setup retry policy, but not satisfy retry condition eg: circuit, retry times == 0, chain stop, ddl
	// failurePolicy DDLStop rpcTimeOut
	failurePolicy := NewFailurePolicy()
	failurePolicy.WithDDLStop()
	RegisterDDLStop(func(ctx context.Context, policy StopPolicy) (bool, string) {
		return true, "TestDDLStop"
	})
	err = rc.Init(map[string]Policy{Wildcard: {
		Enable:        true,
		Type:          0,
		FailurePolicy: failurePolicy,
	}}, nil)
	test.Assert(t, err == nil, err)
	atomic.StoreInt32(&callTimes, 0)
	ri = genRPCInfo()
	ctx = rpcinfo.NewCtxWithRPCInfo(ctx, ri)
	_, ok, err = rc.WithRetryIfNeeded(ctx, &Policy{}, func(ctx context.Context, r Retryer, request, response interface{}) (rpcinfo.RPCInfo, error) {
		atomic.AddInt32(&callTimes, 1)
		if atomic.LoadInt32(&callTimes) == 1 {
			return ri, kerrors.ErrRPCTimeout
		}
		return ri, nil
	}, ri, nil, nil)
	test.Assert(t, kerrors.IsTimeoutError(err), err)
	test.Assert(t, atomic.LoadInt32(&callTimes) == 1)
	test.Assert(t, !ok)

	// backupPolicy MaxRetryTimes = 0
	backupPolicy := NewBackupPolicy(20)
	backupPolicy.StopPolicy.MaxRetryTimes = 0
	err = rc.Init(map[string]Policy{Wildcard: {
		Enable:       true,
		Type:         1,
		BackupPolicy: backupPolicy,
	}}, nil)
	test.Assert(t, err == nil, err)
	atomic.StoreInt32(&callTimes, 0)
	ri = genRPCInfo()
	ctx = rpcinfo.NewCtxWithRPCInfo(ctx, ri)
	_, ok, err = rc.WithRetryIfNeeded(ctx, &Policy{}, func(ctx context.Context, r Retryer, request, response interface{}) (rpcinfo.RPCInfo, error) {
		atomic.AddInt32(&callTimes, 1)
		time.Sleep(time.Millisecond * 100)
		return ri, nil
	}, ri, nil, nil)
	test.Assert(t, err == nil, err)
	test.Assert(t, atomic.LoadInt32(&callTimes) == 1)
	test.Assert(t, ok)
}

func retryCall(callTimes *int32, firstRI rpcinfo.RPCInfo, backup bool) RPCCallFunc {
	// prevRI represents a value of rpcinfo.RPCInfo type.
	var prevRI atomic.Value
	return func(ctx context.Context, r Retryer, request, response interface{}) (rpcinfo.RPCInfo, error) {
		currCallTimes := int(atomic.AddInt32(callTimes, 1))
		cRI := firstRI
		if currCallTimes > 1 {
			cRI = genRPCInfoWithFirstStats(firstRI)
			cRI.Stats().Record(ctx, stats.RPCFinish, stats.StatusInfo, "")
			remoteInfo := remoteinfo.AsRemoteInfo(cRI.To())
			remoteInfo.SetInstance(discovery.NewInstance("tcp", "10.20.30.40:8888", 10, nil))
			if prevRI.Load() == nil {
				prevRI.Store(firstRI)
			}
			r.Prepare(ctx, prevRI.Load().(rpcinfo.RPCInfo), cRI)
			prevRI.Store(cRI)
			return cRI, nil
		} else {
			if backup {
				time.Sleep(20 * time.Millisecond)
				return cRI, nil
			} else {
				return cRI, kerrors.ErrRPCTimeout
			}
		}
	}
}

type mockResult struct {
	result    mockResp
	callTimes int32
}

type mockResp struct {
	code int
	msg  string
}

func (r *mockResult) getResult() interface{} {
	return r.result
}

func (r *mockResult) setResult(ret mockResp) {
	r.result = ret
}

func (r *mockResult) setCallTimes(ct int32) {
	r.callTimes = ct
}

func (r *mockResult) getCallTimes() int32 {
	return r.callTimes
}

func TestNewRetryContainerWithOptions(t *testing.T) {
	t.Run("no_option", func(t *testing.T) {
		rc := NewRetryContainer()
		test.Assertf(t, rc.cbContainer.cbSuite != nil, "cb_suite nil")
		test.Assertf(t, rc.cbContainer.cbStat == true, "cb_stat not true")
	})

	t.Run("percentage_limit", func(t *testing.T) {
		rc := NewRetryContainer(WithContainerEnablePercentageLimit())
		test.Assertf(t, rc.cbContainer.enablePercentageLimit == true, "percentage_limit not true")
	})

	t.Run("percentage_limit&&cbOptions", func(t *testing.T) {
		cbSuite := newCBSuite(nil)
		rc := NewRetryContainer(
			WithContainerEnablePercentageLimit(),
			WithContainerCBSuite(cbSuite),
			WithContainerCBControl(cbSuite.ServiceControl()),
			WithContainerCBPanel(cbSuite.ServicePanel()),
		)
		test.Assertf(t, rc.cbContainer.enablePercentageLimit == true, "percentage_limit not true")
		test.Assertf(t, rc.cbContainer.cbSuite != cbSuite, "cbSuite not ignored")
		test.Assertf(t, rc.cbContainer.cbCtl != cbSuite.ServiceControl(), "cbCtl not ignored")
		test.Assertf(t, rc.cbContainer.cbPanel != cbSuite.ServicePanel(), "cbPanel not ignored")
	})

	t.Run("cb_stat", func(t *testing.T) {
		rc := NewRetryContainer(WithContainerCBStat())
		test.Assertf(t, rc.cbContainer.cbStat == true, "cb_stat not true")
	})

	t.Run("cb_suite", func(t *testing.T) {
		cbs := newCBSuite(nil)
		rc := NewRetryContainer(WithContainerCBSuite(cbs))
		test.Assert(t, rc.cbContainer.cbSuite == cbs, "cbSuite expected %p, got %p", cbs, rc.cbContainer.cbSuite)
	})

	t.Run("cb_control&cb_panel", func(t *testing.T) {
		cbs := newCBSuite(nil)
		rc := NewRetryContainer(
			WithContainerCBControl(cbs.ServiceControl()),
			WithContainerCBPanel(cbs.ServicePanel()))
		test.Assert(t, rc.cbContainer.cbCtl == cbs.ServiceControl(), "cbControl not match")
		test.Assert(t, rc.cbContainer.cbPanel == cbs.ServicePanel(), "cbPanel not match")
	})

	t.Run("CBSuiteOptions", func(t *testing.T) {
		f := func(ctx context.Context, request, response interface{}, err error) circuitbreak.ErrorType {
			return circuitbreak.TypeSuccess
		}
		opts := []circuitbreak.CBSuiteOption{
			circuitbreak.WithServiceGetErrorType(f),
		}

		// test
		rc := NewRetryContainer(WithContainerCBSuiteOptions(opts...))
		serviceControl := rc.cbContainer.cbSuite.ServiceControl()

		// check
		test.Assert(t, len(rc.cbContainer.cbSuiteOptions) == 1)
		test.Assert(t, reflect.ValueOf(serviceControl.GetErrorType).Pointer() == reflect.ValueOf(f).Pointer())
	})

	t.Run("CBSuiteOptions&PercentageLimit", func(t *testing.T) {
		f := func(ctx context.Context, request, response interface{}, err error) circuitbreak.ErrorType {
			return circuitbreak.TypeSuccess
		}
		opts := []circuitbreak.CBSuiteOption{
			circuitbreak.WithServiceGetErrorType(f),
		}

		// test
		rc := NewRetryContainer(WithContainerEnablePercentageLimit(), WithContainerCBSuiteOptions(opts...))
		serviceControl := rc.cbContainer.cbSuite.ServiceControl()

		// check
		test.Assert(t, len(rc.cbContainer.cbSuiteOptions) == 1)
		test.Assert(t, reflect.ValueOf(serviceControl.GetErrorType).Pointer() == reflect.ValueOf(f).Pointer())
	})
}

func TestNewRetryContainerWithCBStat(t *testing.T) {
	cbs := newCBSuite(nil)
	rc := NewRetryContainerWithCBStat(cbs.ServiceControl(), cbs.ServicePanel())
	test.Assert(t, rc.cbContainer.cbCtl == cbs.ServiceControl(), "cbControl not match")
	test.Assert(t, rc.cbContainer.cbPanel == cbs.ServicePanel(), "cbPanel not match")
	test.Assertf(t, rc.cbContainer.cbStat == true, "cb_stat not true")
	rc.Close()
}

func TestNewRetryContainerWithCB(t *testing.T) {
	cbs := newCBSuite(nil)
	rc := NewRetryContainerWithCB(cbs.ServiceControl(), cbs.ServicePanel())
	test.Assert(t, rc.cbContainer.cbCtl == cbs.ServiceControl(), "cbControl not match")
	test.Assert(t, rc.cbContainer.cbPanel == cbs.ServicePanel(), "cbPanel not match")
	test.Assertf(t, rc.cbContainer.cbStat == false, "cb_stat not false")
	rc.Close()
}

func TestPrepareRetryContext(t *testing.T) {
	ctx := PrepareRetryContext(context.Background())

	test.Assert(t, *ctx.Value(CtxReqOp).(*int32) == OpNo)
	*ctx.Value(CtxReqOp).(*int32) = OpDone

	test.Assert(t, *ctx.Value(CtxReqOp).(*int32) == OpDone)
}

var gctx context.Context // make sure the return value of PrepareRetryContext escapes

func BenchmarkPrepareRetryContext(b *testing.B) {
	var ctx context.Context
	for i := 0; i < b.N; i++ {
		ctx = PrepareRetryContext(context.Background())
		_ = ctx.Value(CtxReqOp)
	}
	gctx = ctx
}

// v0.11.0 add type 2 - mixed retry, but the old version has bug that parse the data trigger panic
// json is
// {"enable":true,"type":2,
// "mixed_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}}
// }
// actually, the test cases has cover the bug
// this test case just to record the problem, and ignore it in the future
func TestBugfix4InvalidRetry(t *testing.T) {
	// <v0.11.0: newRetryer panic
	// >v0.11.0: newRetryer successful, mixed retry
	t.Run("type=2, mixed retry", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":2,
"mixed_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}}
}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		r, err := NewRetryer(p, nil, nil)
		test.Assert(t, err == nil, err)
		test.Assert(t, r.(*mixedRetryer).enable)
	})

	// expect
	// <v0.11.0: newRetryer failed, type not match
	// >v0.11.0: newRetryer successful, mixed retry
	t.Run("type=2, mixed retry, but policy has empty failure_policy", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":2,
"mixed_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}},
"failure_policy": {}
}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		r, err := NewRetryer(p, nil, nil)
		test.Assert(t, err == nil, err)
		test.Assert(t, r.(*mixedRetryer).enable)
	})

	// expect
	// <v0.11.0: newRetryer failed, type not match
	// >v0.11.0: newRetryer successful, mixed retry
	t.Run("type=2, mixed retry, but policy has failure_policy", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":2,
"mixed_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}},
"failure_policy": {"stop_policy":{"max_retry_times":0,"max_duration_ms":0,"disable_chain_stop": false, "ddl_stop": false,"cb_policy": {"error_rate":0,"min_sample": 0}},"retry_same_node": false}
}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		r, err := NewRetryer(p, nil, nil)
		test.Assert(t, err == nil, err)
		test.Assert(t, r.(*mixedRetryer).enable)
	})

	// expect
	// <v0.11.0: newRetryer failed, panic
	// >v0.11.0: newRetryer failed, type not match
	t.Run("type=3, unknown retry with unknown policy", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":3,
"new_policy":{"retry_delay_ms":10,"stop_policy":{"max_retry_times":2,"max_duration_ms":0,"disable_chain_stop":false,"ddl_stop":false,"cb_policy":{"error_rate":0.1,"min_sample":200}},"backoff_policy":{"backoff_type":"none"},"retry_same_node":false,"extra":"{\"not_retry_for_timeout\":false,\"rpc_retry_code\":{\"all_error_code\":false,\"error_codes\":[103,1204]},\"biz_retry_code\":{\"all_error_code\":false,\"error_codes\":[]}}","extra_struct":{"not_retry_for_timeout":false,"rpc_retry_code":{"all_error_code":false,"error_codes":[103,1204]},"biz_retry_code":{"all_error_code":false,"error_codes":[]}}}}`
		var p Policy
		err := sonici.UnmarshalFromString(jsonRet, &p)
		test.Assert(t, err == nil, err)

		r, err := NewRetryer(p, nil, nil)
		test.Assert(t, err != nil, err)
		test.Assert(t, r == nil)
		test.Assert(t, strings.Contains(err.Error(), "FailurePolicy is nil or retry type not match"))
	})
}
