package nextgenmetering

import (
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/pingcap/metering_sdk/common"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/metrics"
	"github.com/tidbcloud/metering/pkg/common/models"
	"github.com/tidbcloud/metering/pkg/common/utils"
	"go.uber.org/zap"
)

type SKUGenerator struct {
	conf   *config.ServiceConfig
	logger *zap.Logger
}

// NewSKUGenerator creates a new SKU generator instance
func NewSKUGenerator(conf *config.ServiceConfig, logger *zap.Logger) *SKUGenerator {
	return &SKUGenerator{
		conf:   conf,
		logger: logger,
	}
}

const (
	storageUsageData = "data"
	storageUsageRaft = "raft"
)

// ConvertToMeteringValue converts a value from map[string]interface{} to MeteringValue
func (g *SKUGenerator) ConvertToMeteringValue(value interface{}) (*common.MeteringValue, bool) {
	if value == nil {
		return nil, false
	}

	// Case 1: Already a *common.MeteringValue
	if mv, ok := value.(*common.MeteringValue); ok {
		return mv, true
	}

	// Case 2: A map[string]interface{} that needs to be converted
	if valueMap, ok := value.(map[string]interface{}); ok {
		var meteringValue common.MeteringValue

		// Extract value field
		if val, exists := valueMap["value"]; exists {
			switch v := val.(type) {
			case uint64:
				meteringValue.Value = v
			case int64:
				if v >= 0 {
					meteringValue.Value = uint64(v)
				}
			case int:
				if v >= 0 {
					meteringValue.Value = uint64(v)
				}
			case float64:
				if v >= 0 {
					meteringValue.Value = uint64(v)
				}
			default:
				return nil, false
			}
		} else {
			return nil, false
		}

		// Extract unit field
		if unit, exists := valueMap["unit"]; exists {
			if unitStr, ok := unit.(string); ok {
				meteringValue.Unit = unitStr
			} else {
				return nil, false
			}
		} else {
			return nil, false
		}

		return &meteringValue, true
	}

	return nil, false
}

// GenerateResourceManagerSKUs generates SKU records for resource-manager category
func (g *SKUGenerator) GenerateResourceManagerSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, backupMetaData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake
	skipRCU := isPremiumReservedMode(plan, enrichmentData)

	sourceName := models.NextGenSourceNamePD // Default for resource-manager
	if sourceNameVal, ok := dataPoint[models.NextGenMeteringSourceName].(string); ok {
		sourceName = sourceNameVal
	}

	var olapRU, oltpRU uint64
	var olapUnit, oltpUnit string
	if olapRuData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringOLAPRU]); ok {
		olapRU = olapRuData.Value
		olapUnit = olapRuData.Unit
	}
	if oltpRuData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringOLTPRU]); ok {
		oltpRU = oltpRuData.Value
		oltpUnit = oltpRuData.Unit
	}

	unit := olapUnit
	if unit == "" {
		unit = oltpUnit
	}

	ruVersion := ""
	if v, ok := enrichmentData[models.NextGenMetaRUVersion].(string); ok {
		ruVersion = v
	}

	if !skipRCU {
		generateRegularRCU := true

		if ruVersion == "2" {
			tidbRUV2Data, hasTiDBRUV2 := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringTiDBRUV2])
			tikvRUV2Data, hasTiKVRUV2 := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringTiKVRUV2])
			tiflashRUV2Data, hasTiFlashRUV2 := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringTiFlashRUV2])

			hasAnyRUV2 := hasTiDBRUV2 || hasTiKVRUV2 || hasTiFlashRUV2
			if !hasAnyRUV2 {
				metrics.MetricsNextGenMeteringSkipCount.WithLabelValues(g.conf.Vendor, models.NextGenCategoryResourceManager, constants.MetricsNextGenReasonRUV2Missing).Inc()
				if g.logger != nil {
					g.logger.Warn("ru_v2 metrics all missing when ru_version=2, skipping v2 generation",
						zap.String("cluster_id", clusterID),
						zap.String("shared_pool_id", sharedPoolID),
						zap.Int64("timestamp", timestamp))
				}
			} else {
				var tidbRUV2, tikvRUV2, tiflashRUV2 uint64
				if hasTiDBRUV2 {
					tidbRUV2 = tidbRUV2Data.Value
				}
				if hasTiKVRUV2 {
					tikvRUV2 = tikvRUV2Data.Value
				}
				if hasTiFlashRUV2 {
					tiflashRUV2 = tiflashRUV2Data.Value
				}
				totalRUV2 := tidbRUV2 + tikvRUV2 + tiflashRUV2

				rcuv2Record, err := g.GenerateRCUV2SKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, totalRUV2, unit, tidbRUV2, tikvRUV2, tiflashRUV2, enrichmentData, now)
				if err != nil {
					return nil, err
				}
				billingRecords = append(billingRecords, rcuv2Record)

				totalRU := olapRU + oltpRU
				archiveRecord, err := g.GenerateArchiveRCUSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, totalRU, unit, olapRU, oltpRU, enrichmentData, now)
				if err != nil {
					return nil, err
				}
				billingRecords = append(billingRecords, archiveRecord)

				generateRegularRCU = false
			}
		}

		if generateRegularRCU {
			totalRU := olapRU + oltpRU
			allRecord, err := g.GenerateRCUSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, totalRU, unit, olapRU, oltpRU, enrichmentData, now)
			if err != nil {
				return nil, err
			}
			billingRecords = append(billingRecords, allRecord)
		}
	} else if g.logger != nil {
		g.logger.Debug("skip rcu generation for reserved plan mode",
			zap.String("cluster_id", clusterID),
			zap.String("shared_pool_id", sharedPoolID),
			zap.Int64("timestamp", timestamp),
			zap.String("plan", plan))
	}

	// Generate data-transfer-cross-az SKU from cross_az_traffic_bytes
	if crossAzData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringCrossAZTraffic]); ok && crossAzData.Value > 0 {
		crossAzRecord, err := g.GenerateDataTransferCrossAZSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, crossAzData.Value, crossAzData.Unit, enrichmentData, now)
		if err != nil {
			return nil, err
		}
		billingRecords = append(billingRecords, crossAzRecord)
	}

	// Generate backup-storage-v2 SKU from incremental_write_byte if backup metadata is available
	if incrementalWriteData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringIncrementalWriteByte]); ok && incrementalWriteData.Value > 0 {
		// Check if backup metadata exists and contains backup-storage-v2 extmeta
		if backupMetaData != nil {
			if extMetaVal, ok := backupMetaData[models.NextGenMetaBackupStorageV2]; ok {
				extMeta := ""
				if extMetaStr, ok := extMetaVal.(string); ok {
					extMeta = extMetaStr
				}

				if extMeta != "" && extMeta != models.NextGenBackupTypeFlex {
					backupRecord, err := g.GenerateBackupStorageV2SKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, incrementalWriteData.Value, incrementalWriteData.Unit, extMeta, enrichmentData, now)
					if err != nil {
						return nil, err
					}
					billingRecords = append(billingRecords, backupRecord)
				}
			}
		}
	}

	return billingRecords, nil
}

// GenerateReservedNodeSKUs generates reserved vCPU SKU records for premium_reserved topology-driven clusters.
func (g *SKUGenerator) GenerateReservedNodeSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	usageData, ok := g.ConvertToMeteringValue(dataPoint["usage"])
	if !ok || usageData == nil || usageData.Value == 0 {
		return []models.BillingReportV2Deltalake{}, nil
	}
	skuPlan := plan
	if isPremiumReservedMode(plan, enrichmentData) {
		skuPlan = models.PlanReserved
	}

	nodeType := "unknown"
	if value, ok := dataPoint[models.NextGenMeteringNodeType].(string); ok {
		trimmed := strings.TrimSpace(value)
		if trimmed != "" {
			nodeType = strings.ToLower(trimmed)
		}
	}

	skuUnit := fmt.Sprintf("%s:%s:%s:%s:all", models.NextGenSKUVCPU, cloudProvider, region, skuPlan)
	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: strings.ToLower(usageData.Unit),
		models.NextGenSKULabelKeyNodeType:   nodeType,
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsJSON, _ := json.Marshal(labelsMap)
	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        uuid.New().String(),
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     strconv.FormatUint(usageData.Value, 10),
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       models.NextGenSKUTypePublic,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	return []models.BillingReportV2Deltalake{record}, nil
}

// GenerateProxySKUs generates SKU records for proxy category
func (g *SKUGenerator) GenerateProxySKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake

	sourceName := models.NextGenSourceNameProxy // Default for proxy
	if sourceNameVal, ok := dataPoint[models.NextGenMeteringSourceName].(string); ok {
		sourceName = sourceNameVal
	}

	// Generate data-transfer-out SKU from outBound_bytes if proxy is old version
	if outBoundData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringOutBoundBytes]); ok && outBoundData.Value > 0 {
		outBoundRecord, err := g.GenerateDataTransferOutSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, "", outBoundData.Value, outBoundData.Unit, enrichmentData, now)
		if err != nil {
			return nil, err
		}
		billingRecords = append(billingRecords, outBoundRecord)
	}
	if outBoundData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringPrivateOutBoundBytes]); ok && outBoundData.Value > 0 {
		outBoundRecord, err := g.GenerateDataTransferOutSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, models.NextGenProxyNetworkTypePrivate, outBoundData.Value, outBoundData.Unit, enrichmentData, now)
		if err != nil {
			return nil, err
		}
		billingRecords = append(billingRecords, outBoundRecord)
	}
	if outBoundData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringPublicOutBoundBytes]); ok && outBoundData.Value > 0 {
		outBoundRecord, err := g.GenerateDataTransferOutSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, models.NextGenProxyNetworkTypePublic, outBoundData.Value, outBoundData.Unit, enrichmentData, now)
		if err != nil {
			return nil, err
		}
		billingRecords = append(billingRecords, outBoundRecord)
	}

	// Generate data-transfer-cross-az SKU from crossZone_bytes
	if crossZoneData, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringCrossZoneBytes]); ok && crossZoneData.Value > 0 {
		crossZoneRecord, err := g.GenerateDataTransferCrossAZSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, crossZoneData.Value, crossZoneData.Unit, enrichmentData, now)
		if err != nil {
			return nil, err
		}
		billingRecords = append(billingRecords, crossZoneRecord)
	}

	return billingRecords, nil
}

// generateRCURecord is the shared implementation for generating RCU/RCUV2 SKU records.
// skuName controls the SKU prefix (e.g. "rcu" or "rcuv2"), oltpLabelKey controls the
// label key used for the OLTP value (e.g. "oltpRU" or "ruV2").
func (g *SKUGenerator) generateRCURecord(skuName, oltpLabelKey string, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, totalRU uint64, ruUnit string, olapRU, oltpRU uint64, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	var skuUnit string
	gate := g.conf.NextGenMeteringConfig.GetFeatureGateFilter(constants.FeatureGateNextGenRCUInstance)
	if gate.IsClusterSwitch(clusterID, timestamp) {
		skuUnit = fmt.Sprintf("%s:%s:%s:%s:all:total:%s", skuName, cloudProvider, region, plan, clusterID)
	} else {
		skuUnit = fmt.Sprintf("%s:%s:%s:%s:all:total", skuName, cloudProvider, region, plan)
	}
	unitValue := strconv.FormatUint(totalRU, 10)
	costID := uuid.New().String()
	sourceUnit := strings.ToLower(ruUnit)

	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: sourceUnit,
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsMap[models.NextGenSKULabelKeyOLAPRU] = strconv.FormatUint(olapRU, 10)
	labelsMap[oltpLabelKey] = strconv.FormatUint(oltpRU, 10)
	minRcu, hasMinRcu := enrichmentData[models.NextGenMetaMinRCU].(string)
	if hasMinRcu {
		labelsMap[models.NextGenSKULabelKeyIntervalMin] = minRcu
	} else {
		labelsMap[models.NextGenSKULabelKeyIntervalMin] = models.NextgenFixMinRU
	}

	maxRcu, hasMaxRcu := enrichmentData[models.NextGenMetaMaxRCU].(string)
	meteringMaxRcu, hasMeteringMaxRcu := enrichmentData[models.NextGenMetaMeteringMaxRCU].(string)

	maxGate := g.conf.NextGenMeteringConfig.GetFeatureGateFilter(constants.FeatureGateNextGenRCUMaxV2)
	if maxGate.IsClusterSwitch(clusterID, timestamp) {
		if hasMeteringMaxRcu {
			labelsMap[models.NextGenSKULabelKeyIntervalMax] = meteringMaxRcu
			g.logger.Info("Using meteringMaxRcu for interval max in RCU SKU", zap.String("meteringMaxRcu", meteringMaxRcu), zap.String("maxRcu", maxRcu), zap.String("clusterID", clusterID))
		} else {
			g.logger.Warn("meteringMaxRcu missing", zap.Any("data", enrichmentData))
			if hasMaxRcu {
				labelsMap[models.NextGenSKULabelKeyIntervalMax] = maxRcu
				metrics.MetricsNextGenMeteringMaxRCUFallbackCount.WithLabelValues(
					g.conf.Vendor,
					models.NextGenCategoryResourceManager,
					constants.MetricsNextGenReasonMeteringMaxRCUMissing,
				).Inc()
				g.logger.Info("Falling back to maxRcu for interval max in RCU SKU", zap.String("maxRcu", maxRcu), zap.String("clusterID", clusterID))
			}
		}
	} else if hasMaxRcu {
		labelsMap[models.NextGenSKULabelKeyIntervalMax] = maxRcu
	}

	labelsMap[models.NextGenSKULabelKeyIntervalValue] = unitValue
	labelsJSON, _ := json.Marshal(labelsMap)

	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID || clusterID == "" {
		skuType = models.NextGenSKUTypePrivate
	}

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     "1",
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       skuType,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	return record, nil
}

// GenerateRCUSKU generates RCU SKU record with combined OLAP and OLTP values
func (g *SKUGenerator) GenerateRCUSKU(clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan, component string, totalRU uint64, ruUnit string, olapRU, oltpRU uint64, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	return g.generateRCURecord(models.NextGenSKURCU, models.NextGenSKULabelKeyOLTPRU, clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, totalRU, ruUnit, olapRU, oltpRU, enrichmentData, now)
}

// GenerateRCUV2SKU generates RCU V2 SKU record with combined tidb/tikv/tiflash RU V2 values.
// It reuses the same "rcu" SKU prefix as V1 and adds V2-specific labels (tidbRUV2, tikvRUV2, tiflashRUV2).
func (g *SKUGenerator) GenerateRCUV2SKU(clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan, component string, totalRUV2 uint64, ruUnit string, tidbRUV2, tikvRUV2, tiflashRUV2 uint64, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	// Use V1's SKU prefix "rcu" and oltpLabelKey placeholder for the base record
	record, err := g.generateRCURecord(models.NextGenSKURCU, models.NextGenSKULabelKeyOLTPRU, clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, totalRUV2, ruUnit, 0, 0, enrichmentData, now)
	if err != nil {
		return record, err
	}

	// Parse existing labels, add V2-specific labels, and re-serialize
	var labelsMap map[string]string
	if err := json.Unmarshal([]byte(record.Labels), &labelsMap); err != nil {
		return record, err
	}

	labelsMap[models.NextGenSKULabelKeyTiDBRUV2] = strconv.FormatUint(tidbRUV2, 10)
	labelsMap[models.NextGenSKULabelKeyTiKVRUV2] = strconv.FormatUint(tikvRUV2, 10)
	labelsMap[models.NextGenSKULabelKeyTiFlashRUV2] = strconv.FormatUint(tiflashRUV2, 10)

	labelsJSON, _ := json.Marshal(labelsMap)
	record.Labels = string(labelsJSON)

	return record, nil
}

func (g *SKUGenerator) GenerateArchiveRCUSKU(clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan, component string, totalRU uint64, ruUnit string, olapRU, oltpRU uint64, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	record, err := g.GenerateRCUSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, component, totalRU, ruUnit, olapRU, oltpRU, enrichmentData, now)
	if err != nil {
		return record, err
	}
	record.SKUType = models.NextGenSKUTypePrivate
	return record, nil
}

// GenerateDataTransferOutSKU generates data-transfer-out SKU record
func (g *SKUGenerator) GenerateDataTransferOutSKU(clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan, component, networktype string, bytes uint64, sourceUnit string, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	var skuUnit string
	if networktype != "" {
		skuUnit = fmt.Sprintf("%s:%s:%s:%s:%s:%s", models.NextGenSKUDataTransferOut, cloudProvider, region, plan, component, networktype)
	} else {
		// Generate SKU: data-transfer-out:{cloud-provider}:{region}:{plan}:{component}
		skuUnit = fmt.Sprintf("%s:%s:%s:%s:%s", models.NextGenSKUDataTransferOut, cloudProvider, region, plan, component)
	}

	unitValue := strconv.FormatUint(bytes, 10)
	costID := uuid.New().String()

	// Convert source unit to lower case
	sourceUnitLower := strings.ToLower(sourceUnit)

	// Create labels map following GenerateRCUSKU pattern
	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: sourceUnitLower,
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}
	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID || clusterID == "" {
		skuType = models.NextGenSKUTypePrivate
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     unitValue,
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       skuType, // SKUType should be "public" or "private"
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	return record, nil
}

// GenerateDataTransferCrossAZSKU generates data-transfer-cross-az SKU record
func (g *SKUGenerator) GenerateDataTransferCrossAZSKU(clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan, component string, bytes uint64, sourceUnit string, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	// Generate SKU: data-transfer-cross-az:{cloud-provider}:{region}:{plan}:{component}
	skuUnit := fmt.Sprintf("%s:%s:%s:%s:%s", models.NextGenSKUDataTransferCrossAZ, cloudProvider, region, plan, component)

	unitValue := strconv.FormatUint(bytes, 10)
	costID := uuid.New().String()

	// Convert source unit to lower case
	sourceUnitLower := strings.ToLower(sourceUnit)

	// Determine SKU type based on cluster ID
	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID || clusterID == "" {
		skuType = models.NextGenSKUTypePrivate
	}

	// Create labels map following GenerateDataTransferOutSKU pattern
	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: sourceUnitLower,
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     unitValue,
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       skuType,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	return record, nil
}

// GenerateBackupStorageSKUs generates backup storage SKUs based on written bytes metrics
func (g *SKUGenerator) GenerateBackupStorageSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake

	writtenBytes, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringWrittenBytes])
	if !ok || writtenBytes == nil || writtenBytes.Value == 0 {
		return billingRecords, nil
	}

	backupType := "auto"
	sourceName := models.NextGenSourceNameDFSStats // Default for DFS stats
	if rawSource, ok := dataPoint[models.NextGenMeteringSourceName].(string); ok {
		trimmed := strings.TrimSpace(rawSource)
		if trimmed != "" {
			sourceName = strings.ToLower(trimmed)
		}
	}

	skuUnit := fmt.Sprintf("%s:%s:%s:%s:%s", models.NextGenSKUBackupStorage, cloudProvider, region, plan, backupType)
	unitValue := strconv.FormatUint(writtenBytes.Value, 10)
	costID := uuid.New().String()
	sourceUnit := strings.ToLower(writtenBytes.Unit)

	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: sourceUnit,
		models.NextGenSKULabelKeyBackupType: backupType,
		models.NextGenSKULabelKeySourceName: sourceName,
	}

	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID {
		skuType = models.NextGenSKUTypePrivate
	}

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     unitValue,
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       skuType,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	billingRecords = append(billingRecords, record)
	return billingRecords, nil
}

// GenerateBackupStorageV2SKU generates backup-storage-v2 SKU record with extmeta from backup category metadata
func (g *SKUGenerator) GenerateBackupStorageV2SKU(clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, bytes uint64, sourceUnit, extMeta string, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	// Generate SKU: backup-storage:{cloud-provider}:{region}:{plan}:{extmeta}
	skuUnit := fmt.Sprintf("%s:%s:%s:%s:%s", models.NextGenSKUBackupStorage, cloudProvider, region, plan, extMeta)

	unitValue := strconv.FormatUint(bytes, 10)
	costID := uuid.New().String()

	// Convert source unit to lower case
	sourceUnitLower := strings.ToLower(sourceUnit)

	// Create labels map
	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: sourceUnitLower,
	}

	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	// Determine SKU type based on cluster ID
	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID || clusterID == "" {
		skuType = models.NextGenSKUTypePrivate
	}

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     unitValue,
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       skuType,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	return record, nil
}

// GenerateManualBackupStorageSKUs generates manual backup storage SKUs from GB usage.
func (g *SKUGenerator) GenerateManualBackupStorageSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake

	backupSizeGB, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringBackupSizeGB])
	if !ok || backupSizeGB == nil || backupSizeGB.Value == 0 {
		return billingRecords, nil
	}

	skuUnit := fmt.Sprintf("%s:%s:%s:%s:%s", models.NextGenSKUBackupStorage, cloudProvider, region, plan, models.NextGenBackupTypeManual)
	unitValue := strconv.FormatUint(backupSizeGB.Value, 10)
	costID := uuid.New().String()
	sourceUnit := strings.ToLower(backupSizeGB.Unit)

	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: sourceUnit,
		models.NextGenSKULabelKeyBackupType: models.NextGenBackupTypeManual,
	}
	if backupID, ok := dataPoint[models.NextGenMeteringBackupID].(string); ok && strings.TrimSpace(backupID) != "" {
		labelsMap[models.NextGenSKULabelKeyBackupID] = backupID
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID || clusterID == "" {
		skuType = models.NextGenSKUTypePrivate
	}

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     unitValue,
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       skuType,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	billingRecords = append(billingRecords, record)
	return billingRecords, nil
}

// GenerateFlexBackupStorageSKUs generates flex-mode backup storage SKUs from GB usage.
func (g *SKUGenerator) GenerateFlexBackupStorageSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake

	backupSizeGB, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringBackupSizeGB])
	if !ok || backupSizeGB == nil || backupSizeGB.Value == 0 {
		return billingRecords, nil
	}

	skuUnit := fmt.Sprintf("%s:%s:%s:%s:%s", models.NextGenSKUBackupStorage, cloudProvider, region, plan, models.NextGenBackupTypeFlex)
	unitValue := strconv.FormatUint(backupSizeGB.Value, 10)
	costID := uuid.New().String()
	sourceUnit := strings.ToLower(backupSizeGB.Unit)

	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit: sourceUnit,
		models.NextGenSKULabelKeyBackupType: models.NextGenBackupTypeFlex,
	}
	if backupID, ok := dataPoint[models.NextGenMeteringBackupID].(string); ok && strings.TrimSpace(backupID) != "" {
		labelsMap[models.NextGenSKULabelKeyBackupID] = backupID
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID || clusterID == "" {
		skuType = models.NextGenSKUTypePrivate
	}

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     unitValue,
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       skuType,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	billingRecords = append(billingRecords, record)
	return billingRecords, nil
}

// GenerateDXFSKUs generates SKU records for DXF category
func (g *SKUGenerator) GenerateDXFSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake

	sourceName := models.NextGenSourceNameDXF // Default for DXF
	if sourceNameVal, ok := dataPoint[models.NextGenMeteringSourceName].(string); ok {
		sourceName = sourceNameVal
	}

	// Extract task_type for SKU generation
	taskType := ""
	if taskTypeVal, ok := dataPoint[models.NextGenMeteringTaskType].(string); ok {
		taskType = taskTypeVal
	}

	// Generate task-ru SKU - cost and unitValue are both 1
	taskRuRecord, err := g.GenerateTaskRUSKU(clusterID, sharedPoolID, tenantID, timestamp, cloudProvider, region, plan, sourceName, taskType, dataPoint, enrichmentData, now)
	if err != nil {
		return nil, err
	}
	billingRecords = append(billingRecords, taskRuRecord)

	return billingRecords, nil
}

// GenerateTaskRUSKU generates task-ru SKU record for DXF
func (g *SKUGenerator) GenerateTaskRUSKU(clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan, component, taskType string, dataPoint map[string]interface{}, enrichmentData map[string]interface{}, now time.Time) (models.BillingReportV2Deltalake, error) {
	// Generate SKU: task-ru:{cloud-provider}:{region}:{plan}:{task-type}
	skuUnit := fmt.Sprintf("%s:%s:%s:%s:%s", models.NextGenSKUTaskRU, cloudProvider, region, plan, taskType)

	costID := uuid.New().String()

	// Create labels map with all DXF fields
	labelsMap := map[string]string{}

	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		labelsMap[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	// Add task_id as label
	if taskID, ok := dataPoint[models.NextGenMeteringTaskID]; ok {
		labelsMap[models.NextGenSKULabelKeyTaskID] = fmt.Sprintf("%v", taskID)
	}

	// Add task_type as label
	if taskType != "" {
		labelsMap[models.NextGenSKULabelKeyTaskType] = taskType
	}

	// Add version as label
	if version, ok := dataPoint[models.NextGenMeteringVersion]; ok {
		labelsMap[models.NextGenSKULabelKeyVersion] = fmt.Sprintf("%v", version)
	}

	// Add get_external_requests as label
	if getExtReq, ok := dataPoint[models.NextGenMeteringGetExtRequests]; ok {
		labelsMap[models.NextGenSKULabelKeyGetExtReq] = fmt.Sprintf("%v", getExtReq)
	}

	// Add put_external_requests as label
	if putExtReq, ok := dataPoint[models.NextGenMeteringPutExtRequests]; ok {
		labelsMap[models.NextGenSKULabelKeyPutExtReq] = fmt.Sprintf("%v", putExtReq)
	}

	// Add read_data_bytes as label
	if readDataBytes, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringReadDataBytes]); ok {
		labelsMap[models.NextGenSKULabelKeyReadBytes] = strconv.FormatUint(readDataBytes.Value, 10)
	}

	// Add write_data_bytes as label
	if writeDataBytes, ok := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringWriteDataBytes]); ok {
		labelsMap[models.NextGenSKULabelKeyWriteBytes] = strconv.FormatUint(writeDataBytes.Value, 10)
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     "1", // Fixed to 1 as required
		Cost:          1,   // Fixed to 1 as required
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       models.NextGenSKUTypePrivate, // Set to private for DXF
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	return record, nil
}

func isStorageEncryptionEnabled(enrichmentData map[string]interface{}) bool {
	if len(enrichmentData) == 0 {
		return false
	}

	rawValue, ok := enrichmentData[models.NextGenMetaStorageEncryptionEnabled]
	if !ok || rawValue == nil {
		return false
	}

	switch value := rawValue.(type) {
	case bool:
		return value
	case string:
		enabled, err := strconv.ParseBool(strings.TrimSpace(value))
		return err == nil && enabled
	default:
		return false
	}
}

func (g *SKUGenerator) shouldUseCMEKStorageSKU(clusterID string, tenantID uint64, timestamp int64, enrichmentData map[string]interface{}) bool {
	if !isStorageEncryptionEnabled(enrichmentData) {
		return false
	}

	gate := g.conf.NextGenMeteringConfig.GetFeatureGateFilter(constants.FeatureGateNextGenCMEKStorageSKU)
	if gate.IsFullSwitch(timestamp) {
		return true
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID && gate.IsClusterSwitch(clusterID, timestamp) {
		return true
	}
	if tenantID != 0 && gate.IsTenantSwitch(strconv.FormatUint(tenantID, 10), timestamp) {
		return true
	}
	if tenantValue, ok := enrichmentData[models.NextGenMetaTenantID].(string); ok && gate.IsTenantSwitch(tenantValue, timestamp) {
		return true
	}

	return false
}

// GenerateStorageSizeSKUs generates storage-size SKUs for row/column data
func (g *SKUGenerator) GenerateStorageSizeSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake

	sourceName := "pd"
	if sourceNameVal, ok := dataPoint[models.NextGenMeteringSourceName].(string); ok {
		sourceName = sourceNameVal
	}

	rowStorage, _ := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringRowStorageSize])
	columnStorage, _ := g.ConvertToMeteringValue(dataPoint[models.NextGenMeteringColumnStorageSize])

	// Get disk type for row storage (dataDiskType)
	rowDiskType := "unknown"
	if enrichmentData != nil {
		if rawDiskType, ok := enrichmentData[models.NextGenMetaDiskType]; ok {
			if diskTypeStr, okStr := rawDiskType.(string); okStr {
				diskTypeStr = strings.TrimSpace(diskTypeStr)
				if diskTypeStr != "" {
					rowDiskType = strings.ToLower(diskTypeStr)
				}
			}
		}
	}

	// Get disk type for column storage (columnDataDiskType)
	columnDiskType := "unknown"
	if enrichmentData != nil {
		if rawColumnDiskType, ok := enrichmentData[models.NextGenMetaColumnDataDiskType]; ok {
			if diskTypeStr, okStr := rawColumnDiskType.(string); okStr {
				diskTypeStr = strings.TrimSpace(diskTypeStr)
				if diskTypeStr != "" {
					columnDiskType = strings.ToLower(diskTypeStr)
				}
			}
		}
	}

	skuType := models.NextGenSKUTypePublic
	if clusterID == models.NextGenMeteringSystemID {
		skuType = models.NextGenSKUTypePrivate
	}

	storageMetric := models.NextGenSKUDiskStorageSize
	if g.shouldUseCMEKStorageSKU(clusterID, tenantID, timestamp, enrichmentData) {
		storageMetric = models.NextGenSKUDiskStorageSizeDualEncryption
	}

	baseLabels := map[string]string{
		models.NextGenSKULabelKeySourceName: sourceName,
	}

	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		baseLabels[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		baseLabels[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}
	type layoutMetric struct {
		metric     string
		value      uint64
		sourceUnit string
		diskType   string
	}

	layoutOrder := make([]string, 0, 2)
	layoutSet := make(map[string]struct{}, 2)
	metricsByLayout := make(map[string][]layoutMetric, 2)

	ensureLayout := func(layout string) {
		if _, exists := layoutSet[layout]; !exists {
			layoutSet[layout] = struct{}{}
			layoutOrder = append(layoutOrder, layout)
		}
		if _, exists := metricsByLayout[layout]; !exists {
			metricsByLayout[layout] = make([]layoutMetric, 0, 3)
		}
	}

	addMetric := func(layout string, metric layoutMetric) {
		if metric.value == 0 {
			return
		}
		ensureLayout(layout)
		metricsByLayout[layout] = append(metricsByLayout[layout], metric)
	}

	addStorageMetrics := func(layout string, mv *common.MeteringValue, diskType string) {
		if mv == nil {
			return
		}
		ensureLayout(layout)
		if mv.Value == 0 {
			return
		}
		// TODO(wjz): column storage doesn't support CMEK SKU currently, will be support at 6.30
		sm := storageMetric
		if layout == "column" {
			sm = models.NextGenSKUDiskStorageSize
		}
		addMetric(layout, layoutMetric{
			metric:     sm,
			value:      mv.Value,
			sourceUnit: strings.ToLower(mv.Unit),
			diskType:   diskType,
		})
	}

	addStorageMetrics("row", rowStorage, rowDiskType)
	addStorageMetrics("column", columnStorage, columnDiskType)

	if len(layoutOrder) == 0 {
		ensureLayout("row")
		ensureLayout("column")
	}

	for _, layout := range layoutOrder {
		for _, metric := range metricsByLayout[layout] {
			// Determine storage tier based on disk type from metric
			storageTier := models.GetStorageTierByDiskType(metric.diskType)

			unit := fmt.Sprintf("%s:%s:%s:%s:%s:%s:%s:%s", metric.metric, cloudProvider, region, plan, layout, storageTier, storageUsageData, metric.diskType)
			labels := make(map[string]string, len(baseLabels)+1)
			for k, v := range baseLabels {
				labels[k] = v
			}
			labels[models.NextGenSKULabelKeySourceUnit] = metric.sourceUnit

			unitValue := strconv.FormatUint(metric.value, 10)
			labelsJSON, _ := json.Marshal(labels)
			record := models.BillingReportV2Deltalake{
				SchemeVersion: "v2",
				TenantID:      tenantID,
				CostID:        uuid.New().String(),
				CostTimestamp: timestamp,
				Unit:          unit,
				UnitValue:     unitValue,
				Cost:          1,
				Labels:        string(labelsJSON),
				CreatedAt:     now.Unix(),
				SKUType:       skuType,
				HourDimension: utils.GenerateHourDimension(now.Unix()),
			}
			billingRecords = append(billingRecords, record)
		}
	}

	return billingRecords, nil
}

// GenerateControlPlaneDiskSKUs generates disk-iops and disk-throughput SKUs reported by control-plane
func (g *SKUGenerator) GenerateControlPlaneDiskSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, _ map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	var billingRecords []models.BillingReportV2Deltalake

	sourceName := "control-plane"
	if rawSource, ok := dataPoint[models.NextGenMeteringSourceName].(string); ok {
		trimmed := strings.TrimSpace(rawSource)
		if trimmed != "" {
			sourceName = strings.ToLower(trimmed)
		}
	}

	skuType := models.NextGenSKUTypePublic

	baseLabels := make(map[string]string)
	if sourceName != "" {
		baseLabels[models.NextGenSKULabelKeySourceName] = sourceName
	}
	if clusterID != "" && clusterID != models.NextGenMeteringSystemID {
		baseLabels[models.NextGenSKULabelKeyClusterID] = clusterID
	}
	if sharedPoolID != "" {
		baseLabels[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	type diskMetrics struct {
		count      *common.MeteringValue
		iops       *common.MeteringValue
		throughput *common.MeteringValue
		diskType   string
	}

	components := map[string]*diskMetrics{}

	ensureComponent := func(component string) *diskMetrics {
		if _, exists := components[component]; !exists {
			components[component] = &diskMetrics{diskType: "unknown"}
		}
		return components[component]
	}

	cloneMeteringValue := func(src *common.MeteringValue) *common.MeteringValue {
		if src == nil {
			return nil
		}
		cloned := *src
		return &cloned
	}

	metricConfigs := []struct {
		key       string
		component string
		kind      string
	}{
		{models.NextGenMeteringTiKVDataDiskCount, "tikv", "count"},
		{models.NextGenMeteringTiKVDataDiskIOPS, "tikv", "iops"},
		{models.NextGenMeteringTiKVDataDiskTP, "tikv", "throughput"},
		{models.NextGenMeteringTiKVLogDiskCount, "tikv-log", "count"},
		{models.NextGenMeteringTiKVLogDiskIOPS, "tikv-log", "iops"},
		{models.NextGenMeteringTiKVLogDiskTP, "tikv-log", "throughput"},
		{models.NextGenMeteringTiFlashComputeDiskCount, "tiflash-compute", "count"},
		{models.NextGenMeteringTiFlashComputeDiskIOPS, "tiflash-compute", "iops"},
		{models.NextGenMeteringTiFlashComputeDiskTP, "tiflash-compute", "throughput"},
		{models.NextGenMeteringTiFlashWriteDiskCount, "tiflash-write", "count"},
		{models.NextGenMeteringTiFlashWriteDiskIOPS, "tiflash-write", "iops"},
		{models.NextGenMeteringTiFlashWriteDiskTP, "tiflash-write", "throughput"},
	}

	for _, cfg := range metricConfigs {
		rawValue, exists := dataPoint[cfg.key]
		if !exists {
			continue
		}
		mv, ok := g.ConvertToMeteringValue(rawValue)
		if !ok || mv == nil || mv.Value == 0 {
			continue
		}
		componentMetrics := ensureComponent(cfg.component)
		switch cfg.kind {
		case "count":
			componentMetrics.count = mv
		case "iops":
			componentMetrics.iops = mv
		case "throughput":
			componentMetrics.throughput = mv
		}
	}

	diskTypeConfigs := []struct {
		key       string
		component string
	}{
		{models.NextGenMeteringTiKVDataDiskType, "tikv"},
		{models.NextGenMeteringTiKVLogDiskType, "tikv-log"},
		{models.NextGenMeteringTiFlashComputeDiskType, "tiflash-compute"},
		{models.NextGenMeteringTiFlashWriteDiskType, "tiflash-write"},
	}

	for _, cfg := range diskTypeConfigs {
		rawValue, exists := dataPoint[cfg.key]
		if !exists {
			continue
		}
		var diskType string
		switch v := rawValue.(type) {
		case string:
			diskType = v
		default:
			diskType = fmt.Sprintf("%v", v)
		}
		diskType = strings.ToLower(strings.TrimSpace(diskType))
		if diskType == "" {
			continue
		}
		componentMetrics := ensureComponent(cfg.component)
		componentMetrics.diskType = diskType
	}

	const (
		componentTiFlash        = "tiflash"
		componentTiFlashCompute = "tiflash-compute"
		componentTiFlashWrite   = "tiflash-write"
	)

	tiflashWriteMetrics := components[componentTiFlashWrite]
	tiflashComputeMetrics := components[componentTiFlashCompute]
	if tiflashComputeMetrics != nil || tiflashWriteMetrics != nil {
		combined := &diskMetrics{diskType: "unknown"}
		if tiflashComputeMetrics != nil {
			combined.count = cloneMeteringValue(tiflashComputeMetrics.count)
			combined.iops = cloneMeteringValue(tiflashComputeMetrics.iops)
			combined.throughput = cloneMeteringValue(tiflashComputeMetrics.throughput)
			if tiflashComputeMetrics.diskType != "" {
				combined.diskType = tiflashComputeMetrics.diskType
			}
		}
		if combined.diskType == "" || combined.diskType == "unknown" {
			if tiflashWriteMetrics != nil && tiflashWriteMetrics.diskType != "" {
				combined.diskType = tiflashWriteMetrics.diskType
			}
		}
		if combined.count == nil && tiflashWriteMetrics != nil {
			combined.count = cloneMeteringValue(tiflashWriteMetrics.count)
		} else if combined.count != nil && tiflashWriteMetrics != nil && tiflashWriteMetrics.count != nil {
			combined.count.Value += tiflashWriteMetrics.count.Value
		}
		components[componentTiFlash] = combined
		delete(components, componentTiFlashCompute)
		delete(components, componentTiFlashWrite)
	}

	for component, metrics := range components {
		if metrics == nil || metrics.count == nil || metrics.count.Value == 0 {
			continue
		}
		cost := int64(metrics.count.Value)

		makeRecord := func(metricName string, mv *common.MeteringValue) {
			if mv == nil || mv.Value == 0 {
				return
			}
			labels := make(map[string]string, len(baseLabels)+2)
			for k, v := range baseLabels {
				labels[k] = v
			}
			labels[models.NextGenSKULabelKeySourceUnit] = strings.ToLower(mv.Unit)
			diskType := metrics.diskType
			if diskType == "" {
				diskType = "unknown"
			}
			labels[models.NextGenSKULabelKeyDiskType] = diskType

			if component == componentTiFlash && tiflashWriteMetrics != nil {
				if tiflashWriteMetrics.diskType != "" {
					labels[models.NextGenSKULabelKeyTiFlashWriteDiskType] = tiflashWriteMetrics.diskType
				}
				if metricName == models.NextGenSKUDiskIOPS {
					if tiflashWriteMetrics.iops != nil && tiflashWriteMetrics.iops.Value > 0 {
						labels[models.NextGenSKULabelKeyTiFlashWriteIOPS] = strconv.FormatUint(tiflashWriteMetrics.iops.Value, 10)
						labels[models.NextGenSKULabelKeyTiFlashWriteIOPSUnit] = strings.ToLower(tiflashWriteMetrics.iops.Unit)
					}
				} else if metricName == models.NextGenSKUDiskThroughput {
					if tiflashWriteMetrics.throughput != nil && tiflashWriteMetrics.throughput.Value > 0 {
						labels[models.NextGenSKULabelKeyTiFlashWriteThroughput] = strconv.FormatUint(tiflashWriteMetrics.throughput.Value, 10)
						labels[models.NextGenSKULabelKeyTiFlashWriteThroughputUnit] = strings.ToLower(tiflashWriteMetrics.throughput.Unit)
					}
				}
			}

			// Determine layout based on component
			layout := models.GetStorageLayoutByComponent(component)

			usage := storageUsageData
			if component == "tikv-log" {
				usage = storageUsageRaft
			}

			// Determine storage tier based on disk type
			storageTier := models.GetStorageTierByDiskType(diskType)

			// SKU format: {metric}:{provider}:{region}:{plan}:{layout}:{tier}:{usage}:{diskType}
			unit := fmt.Sprintf("%s:%s:%s:%s:%s:%s:%s:%s", metricName, cloudProvider, region, plan, layout, storageTier, usage, diskType)
			labelsJSON, _ := json.Marshal(labels)

			// Determine SKU type: throughput is always private, others follow the default skuType
			recordSKUType := skuType
			if metricName == models.NextGenSKUDiskThroughput {
				recordSKUType = models.NextGenSKUTypePrivate
			}

			record := models.BillingReportV2Deltalake{
				SchemeVersion: "v2",
				TenantID:      tenantID,
				CostID:        uuid.New().String(),
				CostTimestamp: timestamp,
				Unit:          unit,
				UnitValue:     strconv.FormatUint(mv.Value, 10),
				Cost:          cost,
				Labels:        string(labelsJSON),
				CreatedAt:     now.Unix(),
				SKUType:       recordSKUType,
				HourDimension: utils.GenerateHourDimension(now.Unix()),
			}
			billingRecords = append(billingRecords, record)
		}

		makeRecord(models.NextGenSKUDiskIOPS, metrics.iops)
		makeRecord(models.NextGenSKUDiskThroughput, metrics.throughput)
	}

	return billingRecords, nil
}

// GenerateVCPUUsageSKUs generates vCPU usage SKU records for BYOC worker-node category
func (g *SKUGenerator) GenerateVCPUUsageSKUs(dataPoint map[string]interface{}, clusterID, sharedPoolID string, tenantID uint64, timestamp int64, cloudProvider, region, plan string, enrichmentData map[string]interface{}, now time.Time) ([]models.BillingReportV2Deltalake, error) {
	// BYOC worker-node data uses "usage" key instead of "vcpu_usage"
	vcpuData, ok := g.ConvertToMeteringValue(dataPoint["usage"])
	if !ok || vcpuData == nil {
		g.logger.Warn("worker-node data point missing or invalid usage field",
			zap.Int64("timestamp", timestamp),
			zap.String("shared_pool_id", sharedPoolID))
		metrics.MetricsNextGenMeteringSkipCount.WithLabelValues(cloudProvider, models.NextGenCategoryWorkerNode, constants.MetricsNextGenReasonMetadataMissing).Inc()
		return nil, nil
	}

	skuUnit := fmt.Sprintf("%s:%s:%s:%s:total", models.NextGenSKUVCPU, cloudProvider, region, plan)
	unitValue := strconv.FormatUint(vcpuData.Value, 10)
	costID := uuid.New().String()

	labelsMap := map[string]string{
		models.NextGenSKULabelKeySourceUnit:    strings.ToLower(vcpuData.Unit),
		models.NextGenSKULabelKeyIntervalValue: unitValue,
	}
	if sharedPoolID != "" {
		labelsMap[models.NextGenSKULabelKeySharedPoolID] = sharedPoolID
	}

	labelsJSON, _ := json.Marshal(labelsMap)

	record := models.BillingReportV2Deltalake{
		SchemeVersion: "v2",
		TenantID:      tenantID,
		CostID:        costID,
		CostTimestamp: timestamp,
		Unit:          skuUnit,
		UnitValue:     unitValue,
		Cost:          1,
		Labels:        string(labelsJSON),
		CreatedAt:     now.Unix(),
		SKUType:       models.NextGenSKUTypePublic,
		HourDimension: utils.GenerateHourDimension(now.Unix()),
	}

	return []models.BillingReportV2Deltalake{record}, nil
}