package impl

import (
	"context"
	"fmt"
	"time"

	"github.com/google/uuid"
	app_config "github.com/tidbcloud/dataflow-service-ng/app/config"
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

const bizChangefeedMeteringCategory = "biz_changefeed"
const (
	defaultMeteringHour = 1
	hourInMinutes       = 60
)

type BizChangefeedMeteringWorkerImpl struct {
	machineManager ticdc.MachineManager

	bizChangefeedRepo repo.BizChangefeedRepo
	billingRecordRepo repo.BillingRecordRepo

	mgmtSvcCli grpcli.MgmtSvcGrpcClient
}

func NewBizChangefeedMeteringWorker(
	machineManager ticdc.MachineManager,
	bizChangefeedRepo repo.BizChangefeedRepo,
	billingRecordRepo repo.BillingRecordRepo,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
) biz_changefeed.BizChangefeedMeteringWorker {
	return &BizChangefeedMeteringWorkerImpl{
		machineManager:    machineManager,
		bizChangefeedRepo: bizChangefeedRepo,
		billingRecordRepo: billingRecordRepo,
		mgmtSvcCli:        mgmtSvcCli,
	}
}

func (m *BizChangefeedMeteringWorkerImpl) GenBillingRecordsByCluster(ctx context.Context) error {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).Named("BizChangefeedMeteringWorker")
	bizChangefeeds, bizErr := m.bizChangefeedRepo.List(
		ctx, func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
			return qs.OperationStatusNotIn(
				entity.BizChangefeedOperationStatusCreating,
				entity.BizChangefeedOperationStatusDeleting,
				entity.BizChangefeedOperationStatusDeleted)
		})
	if bizErr != nil {
		logger.Error("list biz_changefeeds error", zap.Error(bizErr))
		return bizErr.Chain("list biz_changefeeds")
	}
	if len(bizChangefeeds) == 0 {
		return nil
	}

	clusters, err := m.listClustersByBizChangefeeds(ctx, bizChangefeeds)
	if err != nil {
		logger.Error("list clusters error", zap.Error(err))
		return fmt.Errorf("list clusters error: %w", err)
	}
	specs, err := m.listSpecifications(ctx)
	if err != nil {
		logger.Error("list specifications error", zap.Error(err))
		return fmt.Errorf("list specifications error: %w", err)
	}

	var multiErr error
	cluster2BizChangefeeds := partitionBizChangefeedsByCluster(bizChangefeeds)
	for clusterID, clusterBizChangefeeds := range cluster2BizChangefeeds {
		cluster, ok := clusters[clusterID]
		if !ok {
			logger.Error("cluster not found", zap.Uint64("cluster_id", clusterID))
			multiErr = multierr.Append(multiErr, fmt.Errorf("cluster %d not found", clusterID))
			continue
		}
		if cluster.GetServicePlan() == commonv1.ServicePlan_BYOC {
			logger.Info("ignore byoc cluster metering",
				zap.Uint64("clusterId", clusterID))
			continue
		}
		if err := m.genBillingRecordForOneCluster(
			ctx, clusterBizChangefeeds, cluster, specs, logger); err != nil {
			logger.Error("gen cluster billing record error",
				zap.Uint64("cluster_id", clusterID), zap.Error(err))
			multiErr = multierr.Append(
				multiErr, fmt.Errorf("gen billing record of cluster %d error: %w", clusterID, err))
			continue
		}
	}
	return multiErr
}

func (m *BizChangefeedMeteringWorkerImpl) genBillingRecordForOneCluster(
	ctx context.Context,
	bizChangefeeds []*entity.BizChangefeed,
	cluster *utils.Cluster,
	specs map[string]*domain.TiCDCClusterSpecification,
	logger *zap.Logger,
) error {
	measuredAt := time.Now().Truncate(time.Hour)
	meteringStart := measuredAt.Add(-time.Duration(defaultMeteringHour) * time.Hour)
	latestBillingRecord, bizErr := m.billingRecordRepo.Get(
		ctx, func(qs entity.BillingRecordQuerySet) entity.BillingRecordQuerySet {
			return qs.ClusterIDEq(entity.ID(cluster.GetClusterID())).
				CategoryEq(bizChangefeedMeteringCategory).
				OrderDescByMeasuredTS()
		})
	if bizErr != nil && bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return bizErr.Chain("get latest billing record")
	}
	if bizErr == nil {
		if !measuredAt.After(latestBillingRecord.MeasuredTS) {
			logger.Info("billing record already exists",
				zap.Uint64("cluster_id", cluster.GetClusterID()),
				zap.String("cost_id", latestBillingRecord.CostID))
			return nil
		}
		meteringStart = latestBillingRecord.MeasuredTS
	}
	// If no billing record exists, meteringStart defaults to defaultMeteringHour ago.
	// genBillingRecord will then meter each changefeed from the later of meteringStart
	// or the changefeed's creation time (truncated to the hour), to avoid double counting.
	billingRecord, err := m.genBillingRecord(bizChangefeeds, meteringStart, measuredAt, cluster, specs)
	if err != nil {
		return fmt.Errorf("gen billing record error: %w", err)
	}
	if bizErr := m.billingRecordRepo.Save(ctx, billingRecord); bizErr != nil {
		return bizErr.Chain("save billing record")
	}
	return nil
}

func (m *BizChangefeedMeteringWorkerImpl) genBillingRecord(
	bizChangefeeds []*entity.BizChangefeed,
	meteringStart time.Time,
	measuredAt time.Time,
	cluster *utils.Cluster,
	specs map[string]*domain.TiCDCClusterSpecification,
) (*entity.BillingRecord, error) {
	cost, err := m.getClusterTotalCost(bizChangefeeds, specs, meteringStart, measuredAt)
	if err != nil {
		return nil, fmt.Errorf("get cluster total cost error: %w", err)
	}
	sku := genChangefeedSKU(cluster.GetCloudProvider(), cluster.GetRegionName(), cluster.GetServicePlan())
	billingRecord := &entity.BillingRecord{
		TenantID:  entity.ID(cluster.GetTenantID()),
		ClusterID: entity.ID(cluster.GetClusterID()),
		Category:  bizChangefeedMeteringCategory,
		ExtraLabels: map[string]string{
			"cluster": cluster.GetClusterName(),
		},
		MeasuredTS: measuredAt,
		CostID:     uuid.NewString(),
		Total:      int64(cost),
		SKU:        sku,
		Synced:     false,
	}
	return billingRecord, nil
}

func (m *BizChangefeedMeteringWorkerImpl) getClusterTotalCost(
	bizChangefeeds []*entity.BizChangefeed,
	specs map[string]*domain.TiCDCClusterSpecification,
	meteringStart time.Time,
	measuredAt time.Time,
) (int, error) {
	if !measuredAt.After(meteringStart) {
		return 0, nil
	}
	var totalCost int
	for _, bizChangefeed := range bizChangefeeds {
		spec, ok := specs[bizChangefeed.SpecificationCode]
		if !ok {
			return 0, fmt.Errorf(
				"specification code %s of biz_changefeed %d is not found",
				bizChangefeed.SpecificationCode, bizChangefeed.ID)
		}
		totalCost += m.getBizChangefeedCost(bizChangefeed, spec, meteringStart, measuredAt)
	}
	return totalCost, nil
}

func (m *BizChangefeedMeteringWorkerImpl) getBizChangefeedCost(
	bizChangefeed *entity.BizChangefeed,
	spec *domain.TiCDCClusterSpecification,
	meteringStart time.Time,
	measuredAt time.Time,
) int {
	effectiveStart := meteringStart
	createdAt := bizChangefeed.CreatedAt.Truncate(time.Hour)
	if createdAt.After(effectiveStart) {
		effectiveStart = createdAt
	}
	if !measuredAt.After(effectiveStart) {
		return 0
	}
	hours := int(measuredAt.Sub(effectiveStart).Hours())
	if hours <= 0 {
		return 0
	}
	return int(spec.GetRCUNumber()) * hours * hourInMinutes
}

func (m *BizChangefeedMeteringWorkerImpl) getClusterTotalRCU(
	bizChangefeeds []*entity.BizChangefeed, specs map[string]*domain.TiCDCClusterSpecification,
) (uint32, error) {
	var totalRCU uint32
	for _, bizChangefeed := range bizChangefeeds {
		spec, ok := specs[bizChangefeed.SpecificationCode]
		if !ok {
			return 0, fmt.Errorf(
				"specification code %s of biz_changefeed %d is not found",
				bizChangefeed.SpecificationCode, bizChangefeed.ID)
		}
		totalRCU += spec.GetRCUNumber()
	}
	return totalRCU, nil
}

func (m *BizChangefeedMeteringWorkerImpl) listClustersByBizChangefeeds(
	ctx context.Context, bizChangefeeds []entity.BizChangefeed,
) (map[uint64]*utils.Cluster, error) {
	clusters := make(map[uint64]*utils.Cluster, len(bizChangefeeds))
	clusterIDS := getClusterIDS(bizChangefeeds)
	listClustersResp, bizErr := m.mgmtSvcCli.ListClusters(ctx, &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			ClusterIds: clusterIDS,
		},
	})
	if bizErr != nil {
		return nil, bizErr.Chain("list clusters, cluster_id: %v", clusterIDS)
	}

	for _, cluster := range listClustersResp.Clusters {
		cw := utils.NewCluster(cluster)
		clusters[cw.GetClusterID()] = cw
	}
	return clusters, nil
}

func (m *BizChangefeedMeteringWorkerImpl) listSpecifications(
	ctx context.Context,
) (map[string]*domain.TiCDCClusterSpecification, error) {
	resp, bizErr := m.machineManager.ListSpecifications(ctx, &domain.ListSpecificationsReq{})
	if bizErr != nil {
		return nil, bizErr
	}

	specs := make(map[string]*domain.TiCDCClusterSpecification, len(resp.Specifications))
	for _, spec := range resp.Specifications {
		specs[spec.Code] = spec
	}
	return specs, nil
}

func getClusterIDS(bizChangefeeds []entity.BizChangefeed) []uint64 {
	clusterIDMap := make(map[uint64]struct{}, len(bizChangefeeds))
	clusterIDS := make([]uint64, 0, len(bizChangefeeds))
	for i := range bizChangefeeds {
		clusterID := bizChangefeeds[i].ClusterID.Uint64()
		if _, ok := clusterIDMap[clusterID]; ok {
			continue
		}

		clusterIDMap[clusterID] = struct{}{}
		clusterIDS = append(clusterIDS, clusterID)
	}
	return clusterIDS
}

func partitionBizChangefeedsByCluster(bizChangefeeds []entity.BizChangefeed) map[uint64][]*entity.BizChangefeed {
	cluster2BizChangefeeds := make(map[uint64][]*entity.BizChangefeed, len(bizChangefeeds))
	for i := range bizChangefeeds {
		clusterID := bizChangefeeds[i].ClusterID.Uint64()
		cluster2BizChangefeeds[clusterID] = append(cluster2BizChangefeeds[clusterID], &bizChangefeeds[i])
	}
	return cluster2BizChangefeeds
}

func genChangefeedSKU(cloudProvider, region string, servicePlan commonv1.ServicePlan) string {
	sku := fmt.Sprintf(app_config.TCUCostUnitPattern, cloudProvider, region)
	if servicePlan == commonv1.ServicePlan_Essential_V2 {
		sku += ":essential"
	}
	return sku
}