package workers

import (
	"errors"
	"fmt"
	"slices"

	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/forms"
	"github.com/pingcap/dbaas/billing2/model/payment/spend_limit"
	"github.com/pingcap/dbaas/billing2/utils/concurrency"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

const spendLimitCheckBatchSize = 500

type spendLimitCheckFn func(tenantID uint, spendLimitSettingID uint) error

func checkSpendLimitBatch(spendLimits []forms.SpendLimitSetting, lastSpendLimitID uint, checkSpendLimit spendLimitCheckFn) uint {
	for _, spendLimit := range spendLimits {
		// Always advance the pagination cursor so a deleted row cannot stall the worker.
		lastSpendLimitID = spendLimit.SpendLimitSettingID
		err := checkSpendLimit(spendLimit.TenantID, spendLimit.SpendLimitSettingID)
		if err != nil {
			if errors.Is(err, billing_errors.ObjectNotExistsError) {
				log.Warn("ignore deleted spend limit", zap.Uint("tenantID", spendLimit.TenantID), zap.Uint("spendLimitSettingID", spendLimit.SpendLimitSettingID), zap.Error(err))
				continue
			}
			log.Error("failed to check spend limit exceed", zap.Uint("tenantID", spendLimit.TenantID), zap.Uint("spendLimitSettingID", spendLimit.SpendLimitSettingID), zap.Error(err))
		}
	}
	return lastSpendLimitID
}

func CheckSpendLimitByTenant(tenantID uint) {
	notCheckWhitelist := config.Get().GetStringSlice(config.SpendLimitNotCheckWhitelist)
	log.Info("get spend limit not check white list", zap.Any("white_list", notCheckWhitelist))
	if slices.Contains(notCheckWhitelist, fmt.Sprintf("%d", tenantID)) {
		log.Info("tenant hit spend limit not check white list", zap.Uint("tenantID", tenantID))
		return
	}
	model := &spend_limit.SpendLimitModel{}
	var lastSpendLimitID uint
	for {
		spendLimits, err := model.ListTenantSpendLimitSettingsAfterID(tenantID, true, lastSpendLimitID, spendLimitCheckBatchSize)
		if err != nil {
			log.Error("failed to get all spend limits in worker.", zap.Error(err))
			return
		}
		if len(spendLimits) == 0 {
			return
		}
		lastSpendLimitID = checkSpendLimitBatch(spendLimits, lastSpendLimitID, model.CheckSpendLimit)
		if len(spendLimits) < spendLimitCheckBatchSize {
			return
		}
	}
}

func CheckAllSpendLimit() error {
	return concurrency.RunInPoolWithAllTenants(
		config.Get().GetUint(config.CheckSpendLimitWorkerConcurrency),
		CheckSpendLimitByTenant,
	)
}