/*
 * 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"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/cloudwego/kitex/pkg/circuitbreak"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/pkg/klog"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
)

func newFailureRetryer(policy Policy, r *ShouldResultRetry, cbC *cbContainer) (Retryer, error) {
	fr := &failureRetryer{failureCommon: &failureCommon{specifiedResultRetry: r, cbContainer: cbC}}
	if err := fr.UpdatePolicy(policy); err != nil {
		return nil, fmt.Errorf("newfailureRetryer failed, err=%w", err)
	}
	return fr, nil
}

type failureRetryer struct {
	enable bool
	*failureCommon
	policy *FailurePolicy
	sync.RWMutex
	errMsg string
}

// ShouldRetry to check if retry request can be called, it is checked in retryer.Do.
// If not satisfy will return the reason message
func (r *failureRetryer) ShouldRetry(ctx context.Context, err error, callTimes int, req interface{}, cbKey string) (string, bool) {
	r.RLock()
	defer r.RUnlock()
	if !r.enable {
		return "", false
	}
	return r.shouldRetry(ctx, callTimes, req, cbKey, r.policy)
}

// AllowRetry implements the Retryer interface.
func (r *failureRetryer) AllowRetry(ctx context.Context) (string, bool) {
	r.RLock()
	defer r.RUnlock()
	if !r.enable || r.policy.StopPolicy.MaxRetryTimes == 0 {
		return "", false
	}
	if stop, msg := chainStop(ctx, r.policy.StopPolicy); stop {
		return msg, false
	}
	if stop, msg := ddlStop(ctx, r.policy.StopPolicy); stop {
		return msg, false
	}
	return "", true
}

// Do implement the Retryer interface.
func (r *failureRetryer) Do(ctx context.Context, rpcCall RPCCallFunc, firstRI rpcinfo.RPCInfo, req, resp interface{}) (lastRI rpcinfo.RPCInfo, recycleRI bool, err error) {
	r.RLock()
	var maxDuration time.Duration
	if r.policy.StopPolicy.MaxDurationMS > 0 {
		maxDuration = time.Duration(r.policy.StopPolicy.MaxDurationMS) * time.Millisecond
	}
	retryTimes := r.policy.StopPolicy.MaxRetryTimes
	r.RUnlock()

	var callTimes int32
	var callCosts strings.Builder
	var cRI rpcinfo.RPCInfo
	cbKey, _ := r.cbContainer.cbCtl.GetKey(ctx, req)
	defer func() {
		if panicInfo := recover(); panicInfo != nil {
			err = panicToErr(ctx, panicInfo, firstRI)
		}
	}()
	startTime := time.Now()
	var curResp interface{}
	newRespFunc := getNewRespFunc(firstRI.Invocation().MethodInfo())
	for i := 0; i <= retryTimes; i++ {
		var callStart time.Time
		if i == 0 {
			callStart = startTime
		} else if i > 0 {
			if ret, e := isExceedMaxDuration(ctx, startTime, maxDuration, callTimes); ret {
				err = e
				break
			}
			if msg, ok := r.ShouldRetry(ctx, err, i, req, cbKey); !ok {
				if msg != "" {
					appendMsg := fmt.Sprintf("retried %d, %s", i-1, msg)
					appendErrMsg(err, appendMsg)
				}
				break
			}
			callStart = time.Now()
			callCosts.WriteByte(',')
		}
		callTimes++
		if r.cbContainer.enablePercentageLimit {
			// record stat before call since requests may be slow, making the limiter more accurate
			recordRetryStat(cbKey, r.cbContainer.cbPanel, callTimes)
		}
		curResp = newRespFunc()
		cRI, err = rpcCall(ctx, r, req, curResp)
		callCosts.WriteString(strconv.FormatInt(time.Since(callStart).Microseconds(), 10))

		if !r.cbContainer.enablePercentageLimit && r.cbContainer.cbStat {
			circuitbreak.RecordStat(ctx, req, nil, err, cbKey, r.cbContainer.cbCtl, r.cbContainer.cbPanel)
		}
		if !r.isRetryResult(ctx, cRI, curResp, err, r.policy) {
			break
		}
	}
	shallowCopyResults(curResp, resp)
	recordRetryInfo(cRI, callTimes, callCosts.String())
	if err == nil && callTimes == 1 {
		return cRI, true, nil
	}
	return cRI, false, err
}

// UpdatePolicy implements the Retryer interface.
func (r *failureRetryer) UpdatePolicy(rp Policy) (err error) {
	r.Lock()
	defer r.Unlock()
	if !rp.Enable {
		r.enable = rp.Enable
		return nil
	}
	if rp.FailurePolicy == nil || rp.Type != FailureType {
		err = fmt.Errorf("FailurePolicy is nil or retry type not match(type=%v), cannot do update in failureRetryer", rp.Type)
		r.errMsg = err.Error()
		return err
	}
	if err = checkStopPolicy(&rp.FailurePolicy.StopPolicy, maxFailureRetryTimes, r); err != nil {
		r.errMsg = err.Error()
		return err
	}
	r.enable = rp.Enable
	r.policy = rp.FailurePolicy
	r.setSpecifiedResultRetryIfNeeded(r.specifiedResultRetry, r.policy)
	if bo, e := initBackOff(rp.FailurePolicy.BackOffPolicy); e != nil {
		r.errMsg = fmt.Sprintf("failureRetryer update BackOffPolicy failed, err=%s", e.Error())
		klog.Warnf("KITEX: %s", r.errMsg)
	} else {
		r.backOff = bo
	}
	return nil
}

// AppendErrMsgIfNeeded implements the Retryer interface.
func (r *failureRetryer) AppendErrMsgIfNeeded(ctx context.Context, err error, ri rpcinfo.RPCInfo, msg string) {
	if r.isRetryErr(ctx, err, ri, r.policy) {
		// Add additional reason when retry is not applied.
		appendErrMsg(err, msg)
	}
}

// Prepare implements the Retryer interface.
func (r *failureRetryer) Prepare(ctx context.Context, prevRI, retryRI rpcinfo.RPCInfo) {
	handleRetryInstance(r.policy.RetrySameNode, prevRI, retryRI)
}

// Type implements the Retryer interface.
func (r *failureRetryer) Type() Type {
	return FailureType
}

// Dump implements the Retryer interface.
func (r *failureRetryer) Dump() map[string]interface{} {
	r.RLock()
	defer r.RUnlock()
	dm := make(map[string]interface{})
	dm["enable"] = r.enable
	dm["failure_retry"] = r.policy
	if r.policy != nil {
		dm["specified_result_retry"] = r.dumpSpecifiedResultRetry(*r.policy)
	}
	if r.errMsg != "" {
		dm["err_msg"] = r.errMsg
	}
	return dm
}

type failureCommon struct {
	backOff              BackOff
	specifiedResultRetry *ShouldResultRetry
	cbContainer          *cbContainer
}

func (f *failureCommon) setSpecifiedResultRetryIfNeeded(rr *ShouldResultRetry, fp *FailurePolicy) {
	if rr != nil {
		// save the object specified by client.WithSpecifiedResultRetry(..)
		f.specifiedResultRetry = rr
	}
	if fp != nil {
		if f.specifiedResultRetry != nil {
			// The priority of client.WithSpecifiedResultRetry(..) is higher, so always update it
			// NOTE: client.WithSpecifiedResultRetry(..) will always reject a nil object
			fp.ShouldResultRetry = f.specifiedResultRetry
		}

		// even though rr passed from this func is nil,
		// the Policy may also have ShouldResultRetry from client.WithFailureRetry or callopt.WithRetryPolicy.
		// convertResultRetry is used to convert 'ErrorRetry and RespRetry' to 'ErrorRetryWithCtx and RespRetryWithCtx'
		fp.convertResultRetry()
	}
}

func (r *failureCommon) isRetryErr(ctx context.Context, err error, ri rpcinfo.RPCInfo, fp *FailurePolicy) bool {
	if err == nil {
		return false
	}
	// Logic Notice:
	// some kinds of error cannot be retried, eg: ServiceCircuitBreak.
	// But CircuitBreak has been checked in ShouldRetry, it doesn't need to filter ServiceCircuitBreak.
	// If there are some other specified errors that cannot be retried, it should be filtered here.

	if fp.isRetryForTimeout() && kerrors.IsTimeoutError(err) {
		return true
	}
	if fp.isErrorRetry(ctx, err, ri) {
		return true
	}
	return false
}

func (r *failureCommon) shouldRetry(ctx context.Context, callTimes int, req interface{}, cbKey string, fp *FailurePolicy) (string, bool) {
	if stop, msg := circuitBreakerStop(ctx, fp.StopPolicy, r.cbContainer, req, cbKey); stop {
		return msg, false
	}
	if stop, msg := ddlStop(ctx, fp.StopPolicy); stop {
		return msg, false
	}
	r.backOff.Wait(callTimes)
	return "", true
}

// isRetryResult to check if the result need to do retry
// Version Change Note:
// < v0.11.0 if the last result still failed, then wrap the error as RetryErr
// >= v0.11.0 don't wrap RetryErr.
// Consideration: Wrap as RetryErr will be reflected as a retry error from monitoring, which is not friendly for troubleshooting
func (r *failureCommon) isRetryResult(ctx context.Context, cRI rpcinfo.RPCInfo, resp interface{}, err error, fp *FailurePolicy) bool {
	if err == nil {
		if fp.isRespRetry(ctx, resp, cRI) {
			// user specified resp to do retry
			return true
		}
	} else if r.isRetryErr(ctx, err, cRI, fp) {
		return true
	}
	return false
}

func (r *failureCommon) dumpSpecifiedResultRetry(fp FailurePolicy) map[string]bool {
	return map[string]bool{
		"error_retry": fp.isErrorRetryWithCtxNonNil(),
		"resp_retry":  fp.isRespRetryWithCtxNonNil(),
		// keep it for some versions to confirm the correctness when troubleshooting
		"old_error_retry": fp.isErrorRetryNonNil(),
		"old_resp_retry":  fp.isRespRetryNonNil(),
	}
}

func initBackOff(policy *BackOffPolicy) (bo BackOff, err error) {
	bo = NoneBackOff
	if policy == nil {
		return
	}
	switch policy.BackOffType {
	case NoneBackOffType:
	case FixedBackOffType:
		if policy.CfgItems == nil {
			return bo, errors.New("invalid FixedBackOff, CfgItems is nil")
		}
		fixMS := policy.CfgItems[FixMSBackOffCfgKey]
		fixMSInt, _ := strconv.Atoi(fmt.Sprintf("%1.0f", fixMS))
		if err = checkFixedBackOff(fixMSInt); err != nil {
			return
		}
		bo = newFixedBackOff(fixMSInt)
	case RandomBackOffType:
		if policy.CfgItems == nil {
			return bo, errors.New("invalid FixedBackOff, CfgItems is nil")
		}
		minMS := policy.CfgItems[MinMSBackOffCfgKey]
		maxMS := policy.CfgItems[MaxMSBackOffCfgKey]
		minMSInt, _ := strconv.Atoi(fmt.Sprintf("%1.0f", minMS))
		maxMSInt, _ := strconv.Atoi(fmt.Sprintf("%1.0f", maxMS))
		if err = checkRandomBackOff(minMSInt, maxMSInt); err != nil {
			return
		}
		bo = newRandomBackOff(minMSInt, maxMSInt)
	default:
		return bo, fmt.Errorf("invalid backoffType=%v", policy.BackOffType)
	}
	return
}

func isExceedMaxDuration(ctx context.Context, start time.Time, maxDuration time.Duration, callTimes int32) (bool, error) {
	if maxDuration > 0 && time.Since(start) > maxDuration {
		return true, makeRetryErr(ctx, fmt.Sprintf("exceed max duration[%v]", maxDuration), callTimes)
	}
	return false, nil
}
