/*
 * 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"
	"sync/atomic"
	"testing"
	"time"

	"github.com/cloudwego/thriftgo/pkg/test"

	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/stats"
)

// test BackupPolicy string
func TestBackupPolicy_String(t *testing.T) {
	r := NewBackupPolicy(100)
	r.WithRetryBreaker(0.3)
	r.WithRetrySameNode()
	msg := "{RetryDelayMS:100 StopPolicy:{MaxRetryTimes:1 MaxDurationMS:0 " +
		"DisableChainStop:false DDLStop:false CBPolicy:{ErrorRate:0.3}} RetrySameNode:true}"
	test.Assert(t, r.String() == msg)
}

// test BackupPolicy call while rpcTime > delayTime
func TestBackupPolicyCall(t *testing.T) {
	ctx := context.Background()
	rc := NewRetryContainer()
	err := rc.Init(map[string]Policy{Wildcard: {
		Enable: true,
		Type:   1,
		BackupPolicy: &BackupPolicy{
			RetryDelayMS: 30,
			StopPolicy: StopPolicy{
				MaxRetryTimes:    2,
				DisableChainStop: false,
				CBPolicy: CBPolicy{
					ErrorRate: defaultCBErrRate,
				},
			},
		},
	}}, nil)
	test.Assert(t, err == nil, err)

	callTimes := int32(0)
	firstRI := genRPCInfo()
	secondRI := genRPCInfoWithRemoteTag(remoteTags)
	ctx = rpcinfo.NewCtxWithRPCInfo(ctx, firstRI)
	ri, ok, err := rc.WithRetryIfNeeded(ctx, &Policy{}, func(ctx context.Context, retryer Retryer, request, response interface{}) (rpcinfo rpcinfo.RPCInfo, err error) {
		atomic.AddInt32(&callTimes, 1)
		if atomic.LoadInt32(&callTimes) == 1 {
			// mock timeout for the first request and get the response of the backup request.
			time.Sleep(time.Millisecond * 50)
			return firstRI, nil
		}
		return secondRI, nil
	}, firstRI, nil, nil)
	test.Assert(t, err == nil, err)
	test.Assert(t, atomic.LoadInt32(&callTimes) == 2)
	test.Assert(t, !ok)
	v, ok := ri.To().Tag(remoteTagKey)
	test.Assert(t, ok)
	test.Assert(t, v == remoteTagValue)
}

func TestBackupRetryWithRPCInfo(t *testing.T) {
	// backup retry
	ctx := context.Background()
	rc := NewRetryContainer()

	ri := genRPCInfo()
	ctx = rpcinfo.NewCtxWithRPCInfo(ctx, ri)
	rpcinfo.Record(ctx, ri, stats.RPCStart, nil)

	// call with retry policy
	var callTimes int32
	policy := BuildBackupRequest(NewBackupPolicy(10))
	ri, ok, err := rc.WithRetryIfNeeded(ctx, &policy, retryCall(&callTimes, ri, true), ri, nil, nil)
	test.Assert(t, err == nil, err)
	test.Assert(t, !ok)
	test.Assert(t, ri.Stats().GetEvent(stats.RPCStart).Status() == stats.StatusInfo)
	test.Assert(t, ri.Stats().GetEvent(stats.RPCFinish).Status() == stats.StatusInfo)
	test.Assert(t, ri.To().Address().String() == "10.20.30.40:8888")
	test.Assert(t, atomic.LoadInt32(&callTimes) == 2)
}

func TestNewRetryer4BackupRequest(t *testing.T) {
	t.Run("normal backup request - success", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":1,
"backup_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.(*backupRetryer).enable)
	})

	t.Run("type is mixedRetry but policy is empty - failed", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":2, "backup_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 == nil)
	})

	t.Run("type is backup but no policy - failed", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":1}`
		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)
	})

	t.Run("type is failure retry but policy is backupRequest - failed", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":0,
"backup_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)
	})

	t.Run("type is backup but has multi-policies - success", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":1,
"backup_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":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.(*backupRetryer).enable)
	})

	t.Run("type is backup but policy json has extra configuration - success", func(t *testing.T) {
		jsonRet := `{"enable":true,"type":1,
"backup_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.(*backupRetryer).enable)
	})
}
