/*
 * 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"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

var errUnexpectedFinish = errors.New("backup request: all retries finished unexpectedly, " +
	"please submit an issue to https://github.com/cloudwego/kitex/issues")

func newBackupRetryer(policy Policy, cbC *cbContainer) (Retryer, error) {
	br := &backupRetryer{cbContainer: cbC}
	if err := br.UpdatePolicy(policy); err != nil {
		return nil, fmt.Errorf("newBackupRetryer failed, err=%w", err)
	}
	return br, nil
}

type backupRetryer struct {
	enable      bool
	retryDelay  time.Duration
	policy      *BackupPolicy
	cbContainer *cbContainer
	sync.RWMutex
	errMsg string
}

type resultWrapper struct {
	ri   rpcinfo.RPCInfo
	resp interface{}
	err  error
}

// ShouldRetry implements the Retryer interface.
func (r *backupRetryer) ShouldRetry(ctx context.Context, err error, callTimes int, req interface{}, cbKey string) (string, bool) {
	r.RLock()
	defer r.RUnlock()
	if !r.enable {
		return "", false
	}
	if stop, msg := circuitBreakerStop(ctx, r.policy.StopPolicy, r.cbContainer, req, cbKey); stop {
		return msg, false
	}
	return "", true
}

// AllowRetry implements the Retryer interface.
func (r *backupRetryer) 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
	}
	return "", true
}

// Do implement the Retryer interface.
func (r *backupRetryer) Do(ctx context.Context, rpcCall RPCCallFunc, firstRI rpcinfo.RPCInfo, req, resp interface{}) (lastRI rpcinfo.RPCInfo, recycleRI bool, err error) {
	r.RLock()
	retryTimes := r.policy.StopPolicy.MaxRetryTimes
	retryDelay := r.retryDelay
	r.RUnlock()

	var callTimes int32 = 0
	var callCosts utils.StringBuilder
	callCosts.RawStringBuilder().Grow(32)
	var recordCostDoing int32 = 0
	var abort int32 = 0
	// notice: buff num of chan is very important here, it cannot less than call times, or the below chan receive will block
	done := make(chan resultWrapper, retryTimes+1)
	cbKey, _ := r.cbContainer.cbCtl.GetKey(ctx, req)
	timer := time.NewTimer(retryDelay)
	defer func() {
		if panicInfo := recover(); panicInfo != nil {
			err = panicToErr(ctx, panicInfo, firstRI)
		}
		timer.Stop()
	}()
	newRespFunc := getNewRespFunc(firstRI.Invocation().MethodInfo())
	// include first call, max loop is retryTimes + 1
	doCall := true
	for i := 0; ; {
		if doCall {
			doCall = false
			i++
			gofunc.GoFunc(ctx, func() {
				if atomic.LoadInt32(&abort) == 1 {
					return
				}
				var (
					e       error
					cRI     rpcinfo.RPCInfo
					curResp = newRespFunc()
				)
				defer func() {
					if panicInfo := recover(); panicInfo != nil {
						e = panicToErr(ctx, panicInfo, firstRI)
					}
					done <- resultWrapper{ri: cRI, resp: curResp, err: e}
				}()
				ct := atomic.AddInt32(&callTimes, 1)
				callStart := time.Now()
				if r.cbContainer.enablePercentageLimit {
					// record stat before call since requests may be slow, making the limiter more accurate
					recordRetryStat(cbKey, r.cbContainer.cbPanel, ct)
				}
				cRI, e = rpcCall(ctx, r, req, curResp)
				recordCost(ct, callStart, &recordCostDoing, &callCosts, &abort, e)
				if !r.cbContainer.enablePercentageLimit && r.cbContainer.cbStat {
					circuitbreak.RecordStat(ctx, req, nil, e, cbKey, r.cbContainer.cbCtl, r.cbContainer.cbPanel)
				}
			})
		}
		select {
		case <-timer.C:
			if _, ok := r.ShouldRetry(ctx, nil, i, req, cbKey); ok && i <= retryTimes {
				doCall = true
				timer.Reset(retryDelay)
			}
		case res := <-done:
			shallowCopyResults(res.resp, resp)
			atomic.StoreInt32(&abort, 1)
			recordRetryInfo(res.ri, atomic.LoadInt32(&callTimes), callCosts.String())
			return res.ri, false, res.err
		}
	}
}

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

// UpdatePolicy implements the Retryer interface.
func (r *backupRetryer) UpdatePolicy(rp Policy) (err error) {
	r.Lock()
	defer r.Unlock()
	if !rp.Enable {
		r.enable = rp.Enable
		return nil
	}
	if rp.BackupPolicy == nil || rp.Type != BackupType {
		err = fmt.Errorf("BackupPolicy is nil or retry type not match(type=%v), cannot do update in backupRetryer", rp.Type)
		r.errMsg = err.Error()
		return err
	}
	if rp.BackupPolicy.RetryDelayMS == 0 {
		err = errors.New("invalid retry delay duration in backupRetryer")
		r.errMsg = err.Error()
		return err
	}
	if err = checkStopPolicy(&rp.BackupPolicy.StopPolicy, maxBackupRetryTimes, r); err != nil {
		r.errMsg = err.Error()
		return err
	}
	r.enable = rp.Enable
	r.policy = rp.BackupPolicy
	r.retryDelay = time.Duration(rp.BackupPolicy.RetryDelayMS) * time.Millisecond
	return nil
}

// AppendErrMsgIfNeeded implements the Retryer interface.
func (r *backupRetryer) AppendErrMsgIfNeeded(ctx context.Context, err error, ri rpcinfo.RPCInfo, msg string) {
	if kerrors.IsTimeoutError(err) {
		// Add additional reason to the error message when timeout occurs but the backup request is not sent.
		appendErrMsg(err, msg)
	}
}

// Dump implements the Retryer interface.
func (r *backupRetryer) Dump() map[string]interface{} {
	r.RLock()
	defer r.RUnlock()
	dm := map[string]interface{}{"enable": r.enable, "backup_request": r.policy}
	if r.errMsg != "" {
		dm["err_msg"] = r.errMsg
	}
	return dm
}

// Type implements the Retryer interface.
func (r *backupRetryer) Type() Type {
	return BackupType
}

// record request cost, it may execute concurrent
func recordCost(ct int32, start time.Time, recordCostDoing *int32, sb *utils.StringBuilder, abort *int32, err error) {
	if atomic.LoadInt32(abort) == 1 {
		return
	}
	for !atomic.CompareAndSwapInt32(recordCostDoing, 0, 1) {
		runtime.Gosched()
	}
	sb.WithLocked(func(b *strings.Builder) error {
		if b.Len() > 0 {
			b.WriteByte(',')
		}
		b.WriteString(strconv.Itoa(int(ct)))
		b.WriteByte('-')
		b.WriteString(strconv.FormatInt(time.Since(start).Microseconds(), 10))
		return nil
	})
	atomic.StoreInt32(recordCostDoing, 0)
}
