package workers

import (
	"errors"
	"time"

	"github.com/avast/retry-go"
	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/db"
	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/handler/centralhandler"
	"github.com/pingcap/dbaas/billing2/model/billing"
	"github.com/pingcap/dbaas/billing2/utils/concurrency"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"github.com/pingcap/dbaas/billing2/utils/timeutils"
	"go.uber.org/zap"
)

func RemoveUnusedCosts(tenantIDs ...uint) error {
	if !config.Get().GetBool(config.EnableRemoveUnusedCosts) {
		log.Info("remove unused costs is disabled")
		return nil
	}
	log.Info("start remove unused costs")
	var err error
	measureModel := &billing.MeasureModel{}
	if len(tenantIDs) == 0 {
		tenants, err := centralhandler.GetCentralHandler().GetAllTenants()
		if err != nil {
			return err
		}
		for _, tenant := range tenants {
			tenantIDs = append(tenantIDs, tenant.TenantID)
		}
	}
	et := getRemoveUnusedCostsEndTime()
	log.Info("remove unused costs, calc end time", zap.Any("end_time", et))

	for _, tenantID := range tenantIDs {
		_ = retry.Do(func() error {
			return measureModel.RemoveUnusedCosts(et, tenantID)
		}, retry.Attempts(3), retry.Delay(time.Millisecond*300), retry.MaxDelay(time.Second*2))

		time.Sleep(time.Millisecond * time.Duration(config.Get().GetInt(config.RemoveUnusedCostsOrgSleepMs)))
	}
	log.Info("end remove unused costs")
	return err
}

func getRemoveUnusedCostsEndTime() time.Time {
	m := config.Get().GetInt(config.RemoveUnusedCostsMonthBefore)
	now := timeutils.GetNow()
	dateTime := now.AddDate(0, -m, 0)

	year, month, _ := dateTime.Date()
	startTimeOfMonth := time.Date(year, month, 1, 0, 0, 0, 0, now.Location())
	endTimeOfLastMonth := startTimeOfMonth.Add(-time.Second)

	return endTimeOfLastMonth
}

func RemoveUnusedRawCosts(tenantIDs ...uint) error {
	var err error
	measureModel := &billing.MeasureModel{}
	dt := timeutils.GetNow().Add(-config.Get().GetDuration(config.RemoveUnusedCostsDays) * time.Hour * 24)
	offset := config.Get().GetDuration(config.RemoveUnusedCostsOffsetSeconds) * time.Second
	if len(tenantIDs) == 0 {
		tenants, err := centralhandler.GetCentralHandler().GetAllTenants()
		if err != nil {
			return err
		}
		for _, tenant := range tenants {
			tenantIDs = append(tenantIDs, tenant.TenantID)
		}
	}
	err = concurrency.RunInPoolWithIDs(
		config.Get().GetUint(config.RemoveUnusedCostsWorkerConcurrency),
		func(tenantID uint) {
			err := retry.Do(func() error {
				return measureModel.RemoveUnusedRawCosts(dt, dt.Add(offset), tenantID)
			})
			if err != nil {
				log.Error("failed to remove unused costs", zap.Error(err), zap.Uint("tenantID", tenantID), zap.Time("dt", dt), zap.Duration("offset", offset))
				return
			}
			log.Info("remove all unused costs", zap.Uint("tenantID", tenantID), zap.Time("dt", dt), zap.Duration("offset", offset))
		},
		tenantIDs,
	)
	return err
}

func MergeSmallCosts(tenantIDs ...uint) error {
	if !config.Get().GetBool(config.EnableOfflineMerging) {
		log.Info("offline merging is disabled")
		return nil
	}
	log.Info("begin merge small costs", zap.Uints("tenantIDs", tenantIDs))
	var err error
	measureModel := &billing.MeasureModel{}
	offset := config.Get().GetDuration(config.MergeSmallCostsOffsetSeconds) * time.Second
	maxOffset := 24 * time.Hour
	if offset > maxOffset {
		err = billing_errors.NewSystemError(errors.New("offset out of maxOffset, it will break the bills"))
		log.Error(err.Error(), zap.Error(err), zap.Duration("offset", offset), zap.Duration("maxOffset", maxOffset))
		return err
	}
	dt := timeutils.GetNow().Add(-config.Get().GetDuration(config.MergeSmallCostsDays) * time.Hour * 24).Truncate(offset)
	if len(tenantIDs) == 0 {
		tenants, err := centralhandler.GetCentralHandler().GetAllTenants()
		if err != nil {
			return err
		}
		for _, tenant := range tenants {
			tenantIDs = append(tenantIDs, tenant.TenantID)
		}
	}
	log.Info("start merge small costs", zap.Uints("tenantIDs", tenantIDs), zap.Time("dt", dt))

	err = concurrency.RunInPoolWithIDs(
		config.Get().GetUint(config.MergeSmallCostsWorkerConcurrency),
		func(tenantID uint) {
			err := retry.Do(func() error {
				return measureModel.MergeSmallUnits(dt, dt.Add(offset), tenantID)
			})
			if err != nil {
				log.Error("failed to merged small costs", zap.Error(err), zap.Uint("tenantID", tenantID), zap.Time("dt", dt), zap.Duration("offset", offset))
				return
			}
			log.Info("merged small costs", zap.Uint("tenantID", tenantID), zap.Time("dt", dt), zap.Duration("offset", offset))
		},
		tenantIDs,
	)
	return err
}

func MarkMergedCostsToRaw(tenantIDs ...uint) error {
	var err error
	offset := config.Get().GetDuration(config.MergeSmallCostsOffsetSeconds) * time.Second
	dt := timeutils.GetNow().Add(-config.Get().GetDuration(config.MergeSmallCostsDays) * time.Hour * 24).Truncate(offset)
	if len(tenantIDs) == 0 {
		tenants, err := centralhandler.GetCentralHandler().GetAllTenants()
		if err != nil {
			return err
		}
		for _, tenant := range tenants {
			tenantIDs = append(tenantIDs, tenant.TenantID)
		}
	}
	err = concurrency.RunInPoolWithIDs(
		config.Get().GetUint(config.MergeSmallCostsWorkerConcurrency),
		func(tenantID uint) {
			num, err := entity.NewUserUnitCostQuerySet(db.GetDB()).
				TenantIDEq(tenantID).CostTimeStampGt(dt.Unix() - 1).CostTimeStampLt(dt.Add(offset).Unix()).
				TypeEq(entity.UserUnitCostType_Merged).GetUpdater().
				SetType(entity.UserUnitCostType_Raw).UpdateNum()
			if err != nil {
				log.Error("failed to mark merged costs to raw", zap.Error(err), zap.Uint("tenantID", tenantID), zap.Time("dt", dt), zap.Duration("offset", offset))
				return
			}
			log.Info("success to mark merged costs to raw", zap.Int64("affectRows", num), zap.Uint("tenantID", tenantID), zap.Time("dt", dt), zap.Duration("offset", offset))
		},
		tenantIDs,
	)
	return err
}