package costexplorer

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/constant"
	"github.com/pingcap/dbaas/billing2/db"
	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/forms"
	"github.com/pingcap/dbaas/billing2/handler/centralhandler"
	"github.com/pingcap/dbaas/billing2/utils/labels"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"github.com/pingcap/dbaas/billing2/utils/timeutils"
	"github.com/shopspring/decimal"
	"go.uber.org/zap"
)

// errCostsDetailServiceUnresolved mirrors the legacy "return nil, nil" early-exit used by the
// *FromUnitCosts functions when GetServiceName fails for a row: the whole result becomes nil.
var errCostsDetailServiceUnresolved = errors.New("costs detail service unresolved")

// costsDetailNameEnricher attaches project/cluster names to a streamed CostsDetailsQuery row,
// reproducing CostsDetailsQueryWithClusterAndProjectName one row at a time. The time budget is
// started lazily on the first row so it excludes the (potentially slow) query execution time.
type costsDetailNameEnricher struct {
	enabled         bool
	disabledTenants []string
	start           time.Time
	expired         bool
}

func newCostsDetailNameEnricher() *costsDetailNameEnricher {
	return &costsDetailNameEnricher{
		enabled:         config.Get().GetBool(config.BillsDisplayProjectAndClusterNameWithID),
		disabledTenants: config.Get().GetStringSlice(config.DisableClusterDisplayTenants),
	}
}

func (e *costsDetailNameEnricher) enrich(cost *entity.CostsDetailsQuery) {
	if !e.enabled || e.expired {
		return
	}
	if e.start.IsZero() {
		e.start = timeutils.GetNow()
	}
	if e.start.Add(time.Second * config.Get().GetDuration(config.MaxAttachLabelTimeSecond)).Before(timeutils.GetNow()) {
		log.Error("failed to attach cluster name and project name to unit costs", zap.Uint("tenantID", cost.TenantID))
		e.expired = true
		return
	}
	cost.Labels = labels.WithProjectName(cost.Labels)
	if !slices.Contains(e.disabledTenants, fmt.Sprint(cost.TenantID)) {
		cost.Labels = labels.WithClusterName(cost.TenantID, cost.Unit, cost.Labels)
	}
}

// streamCostsDetailRows runs an aggregation query against user_unit_costs and feeds each result
// row to handle without buffering the full result set in a Go slice. This is the memory-bounded
// counterpart of the legacy ".Find(&slice)" detail loaders: peak Go memory becomes the size of
// the de-duplicated accumulator output instead of every grouped row.
//
// The SQL (and its ORDER BY) is identical to the legacy queries, so the row stream order — and
// therefore the first-seen ordering of any accumulator output — is byte-for-byte unchanged.
//
// Connection lifetime: project/cluster name enrichment may perform cache/network lookups per row.
// To avoid holding a read-replica connection open for the duration of those lookups, the active
// enrich path drains the DB cursor first (so database/sql releases the connection once rows are
// exhausted) and only then enriches+handles each row. The non-enrich path — which includes the
// big-customer / high-volume case — streams row-by-row with no buffering.
func streamCostsDetailRows(query string, params map[string]interface{}, queryTimeout time.Duration, enrich bool, logTenantIDs []uint, handle func(entity.CostsDetailsQuery) error) error {
	ctx, cancel := context.WithTimeout(context.Background(), queryTimeout)
	defer cancel()

	var enricher *costsDetailNameEnricher
	if enrich {
		enricher = newCostsDetailNameEnricher()
	}
	bufferForEnrich := enricher != nil && enricher.enabled

	readDB := db.GetReadOnlyDB()
	queryStartTime := time.Now()
	rows, err := readDB.WithContext(ctx).Raw(query, params).Rows()
	if err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			log.Error("query timeout for bills details streaming",
				zap.Uints("tenantIDs", logTenantIDs),
				zap.Duration("timeout", queryTimeout),
				zap.Error(err))
			return billing_errors.NewSystemError(fmt.Errorf("query timeout after %v: %w", queryTimeout, err))
		}
		return billing_errors.NewSystemError(err)
	}
	defer rows.Close()

	count := 0
	// buffered only holds rows when name enrichment is actually enabled; it lets us close the
	// cursor (release the connection) before running the per-row enrichment lookups.
	var buffered []entity.CostsDetailsQuery
	for rows.Next() {
		var cost entity.CostsDetailsQuery
		if err := readDB.ScanRows(rows, &cost); err != nil {
			if errors.Is(err, context.DeadlineExceeded) {
				log.Error("query timeout while scanning bills details streaming",
					zap.Uints("tenantIDs", logTenantIDs),
					zap.Duration("timeout", queryTimeout),
					zap.Error(err))
				return billing_errors.NewSystemError(fmt.Errorf("query timeout after %v: %w", queryTimeout, err))
			}
			return billing_errors.NewSystemError(err)
		}
		count++
		if bufferForEnrich {
			buffered = append(buffered, cost)
			continue
		}
		if err := handle(cost); err != nil {
			return err
		}
	}
	if err := rows.Err(); err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			log.Error("query timeout while streaming bills details",
				zap.Uints("tenantIDs", logTenantIDs),
				zap.Duration("timeout", queryTimeout),
				zap.Error(err))
			return billing_errors.NewSystemError(fmt.Errorf("query timeout after %v: %w", queryTimeout, err))
		}
		return billing_errors.NewSystemError(err)
	}
	// The cursor is fully drained here, so database/sql has released the connection back to the
	// pool. Enrichment lookups below no longer hold a DB connection.
	rows.Close()

	queryDuration := time.Since(queryStartTime)
	slowLogThreshold := config.Get().GetDuration(config.DBQuerySlowLogThresholdSeconds) * time.Second
	if queryDuration > slowLogThreshold {
		log.Warn("slow streaming query detected for bills details",
			zap.Int("costsCount", count),
			zap.Uints("tenantIDs", logTenantIDs),
			zap.Duration("queryDuration", queryDuration),
			zap.Duration("timeout", queryTimeout))
	} else {
		log.Info("streaming bills details query done",
			zap.Int("costsCount", count),
			zap.Uints("tenantIDs", logTenantIDs),
			zap.Duration("queryDuration", queryDuration))
	}

	if bufferForEnrich {
		for i := range buffered {
			enricher.enrich(&buffered[i])
			if err := handle(buffered[i]); err != nil {
				return err
			}
		}
	}
	return nil
}

// buildCostsDetailMultiBillingConfigsQuery reproduces the query construction of
// getCostsSliceWithMultiBillingConfigs exactly (same filter order, ORDER BY, force index, and
// big-customer label-drop behavior), so the slice and streaming paths are guaranteed identical.
func buildCostsDetailMultiBillingConfigsQuery(tenantIDs []uint, billingConfigIDs []uint, year int, month int, invoiceIDs []int, clusterID uint) (string, map[string]interface{}, time.Duration, bool) {
	filter := []string{}
	params := map[string]interface{}{}
	queryTime := getCostQueryTime(year, month)

	if len(tenantIDs) != 0 {
		filter = append(filter, "tenant_id in @tenant_ids")
		params["tenant_ids"] = tenantIDs
	}
	if clusterID != 0 {
		filter = append(filter, "cluster_id = @cluster_id")
		params["cluster_id"] = clusterID
	}
	filter = append(filter, "status = @status")
	params["status"] = entity.UserUnitCostStatus_DONE
	filter = append(filter, "cost_time_stamp > @start_date")
	params["start_date"] = queryTime.StartDate.Unix() - 1
	filter = append(filter, "cost_time_stamp < @end_date")
	params["end_date"] = queryTime.EndDate.Unix()
	filter = append(filter, "internal = @internal")
	params["internal"] = false

	if len(billingConfigIDs) != 0 {
		filter = append(filter, "binded_billing_config_id in @binded_billing_config_ids")
		params["binded_billing_config_ids"] = billingConfigIDs
	}

	if len(invoiceIDs) != 0 {
		filter = append(filter, "invoice_id in @invoice_ids")
		params["invoice_ids"] = invoiceIDs
	}

	disabledTenants := config.Get().GetStringSlice(config.DisableClusterDisplayTenants)
	shouldRemoveLabelsFromGroupBy := false

	if clusterID == 0 && len(tenantIDs) > 0 {
		for _, tenantID := range tenantIDs {
			if slices.Contains(disabledTenants, fmt.Sprint(tenantID)) {
				shouldRemoveLabelsFromGroupBy = true
				break
			}
		}

		if shouldRemoveLabelsFromGroupBy {
			cutoffDate := time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC)
			if queryTime.StartDate.Before(cutoffDate) {
				shouldRemoveLabelsFromGroupBy = false
			}
		}
	}

	var query string
	if shouldRemoveLabelsFromGroupBy {
		query = fmt.Sprintf(` select tenant_id, binded_billing_config_id, substr(from_unixtime(cost_time_stamp), 1, 10 ) as billed_date,  unit, consumed_source, sum(cost) as cost, sum(cost_price) as cost_price, sum(raw_price) as raw_price
    from user_unit_costs %s where %s
    group by tenant_id, binded_billing_config_id, unit, consumed_source ,substr(from_unixtime(cost_time_stamp), 1, 10 )    order by  tenant_id, binded_billing_config_id, substr(from_unixtime(cost_time_stamp), 1, 10 ) ;
    `, forceClusterIndex, strings.Join(filter, " and "))
	} else {
		query = fmt.Sprintf(` select tenant_id, binded_billing_config_id, substr(from_unixtime(cost_time_stamp), 1, 10 ) as billed_date,  unit, labels, consumed_source, sum(cost) as cost, sum(cost_price) as cost_price, sum(raw_price) as raw_price
    from user_unit_costs %s where %s
    group by tenant_id, binded_billing_config_id, unit, labels, consumed_source ,substr(from_unixtime(cost_time_stamp), 1, 10 )    order by  tenant_id, binded_billing_config_id, substr(from_unixtime(cost_time_stamp), 1, 10 ) ;
    `, forceClusterIndex, strings.Join(filter, " and "))
	}

	queryTimeout := config.Get().GetDuration(config.PerDBQueryTimeoutSeconds) * time.Second
	return query, params, queryTimeout, shouldRemoveLabelsFromGroupBy
}

// forEachCostsDetailMultiBillingConfigs streams the monthly bills-detail aggregation for the
// given tenants/billing configs without materializing the whole result set.
// Multi-day (monthly) queries are split into per-day sub-queries so each one completes
// within the DB timeout — a full-month GROUP BY on user_unit_costs can exceed 30s.
func forEachCostsDetailMultiBillingConfigs(tenantIDs []uint, billingConfigIDs []uint, year int, month int, invoiceIDs []int, clusterID uint, handle func(entity.CostsDetailsQuery) error) error {
	if len(tenantIDs) == 1 && len(billingConfigIDs) == 1 && len(invoiceIDs) == 0 {
		startDate := timeutils.GetMonthDateByYearMonth(year, month)
		endDate := startDate.AddDate(0, 1, 0)
		if clusterID == 0 {
			return forEachCostsDetailForDateRange(tenantIDs[0], billingConfigIDs[0], startDate, endDate, handle)
		}
		return forEachCostsDetailForDateRangeWithCluster(tenantIDs[0], billingConfigIDs[0], startDate, endDate, clusterID, handle)
	}
	query, params, queryTimeout, shouldRemoveLabelsFromGroupBy := buildCostsDetailMultiBillingConfigsQuery(tenantIDs, billingConfigIDs, year, month, invoiceIDs, clusterID)
	enrich := !shouldRemoveLabelsFromGroupBy

	startDate := timeutils.GetMonthDateByYearMonth(year, month)
	endDate := startDate.AddDate(0, 1, 0)
	days := splitDateRangeByDay(startDate, endDate)
	if len(days) > 1 {
		splitCtx, splitCancel := context.WithTimeout(context.Background(),
			getSplitLoopTimeout(shouldRemoveLabelsFromGroupBy, time.Time{}))
		defer splitCancel()

		for _, day := range days {
			if err := splitCtx.Err(); err != nil {
				return billing_errors.NewSystemError(fmt.Errorf("bills detail query cancelled: %w", err))
			}
			dayParams := copyCostQueryParams(params)
			dayParams["start_date"] = day.start.Unix() - 1
			dayParams["end_date"] = day.end.Unix()
			if err := streamCostsDetailRows(query, dayParams, queryTimeout, enrich, tenantIDs, handle); err != nil {
				return err
			}
		}
		return nil
	}
	return streamCostsDetailRows(query, params, queryTimeout, enrich, tenantIDs, handle)
}

// buildDailyCostsByDateRangeQuery reproduces GetDailyCostsByDateRange(ExcludingUnits) so the
// slice and streaming variants stay identical.
func buildDailyCostsByDateRangeQuery(tenantID uint, billingConfigID uint, startDate, endDate time.Time, excludeUnits []string) (string, map[string]interface{}, time.Duration) {
	filter := []string{
		"tenant_id = @tenant_id",
		"status = @status",
		"cost_time_stamp > @start_date",
		"cost_time_stamp < @end_date",
		"internal = @internal",
	}
	params := map[string]interface{}{
		"tenant_id":  tenantID,
		"status":     entity.UserUnitCostStatus_DONE,
		"start_date": startDate.Unix() - 1,
		"end_date":   endDate.Unix(),
		"internal":   false,
	}
	if billingConfigID != 0 {
		filter = append(filter, "binded_billing_config_id = @binded_billing_config_id")
		params["binded_billing_config_id"] = billingConfigID
	}
	if len(excludeUnits) > 0 {
		filter = append(filter, "unit NOT IN @exclude_units")
		params["exclude_units"] = excludeUnits
	}

	query := fmt.Sprintf(`select tenant_id, binded_billing_config_id, substr(from_unixtime(cost_time_stamp), 1, 10) as billed_date, unit, labels, consumed_source, sum(cost_price) as cost_price, sum(raw_price) as raw_price
    from user_unit_costs %s where %s
    group by tenant_id, binded_billing_config_id, unit, labels, consumed_source, substr(from_unixtime(cost_time_stamp), 1, 10)
    order by substr(from_unixtime(cost_time_stamp), 1, 10);`, forceClusterIndex, strings.Join(filter, " and "))

	queryTimeout := config.Get().GetDuration(config.PerDBQueryTimeoutSeconds) * time.Second
	return query, params, queryTimeout
}

// ForEachDailyCostByDateRangeExcludingUnits streams the daily aggregated costs (optionally
// excluding units) so callers can accumulate row-by-row instead of loading the full slice.
// Multi-day ranges are split into per-day sub-queries to keep each DB query under the timeout.
func (c *CostExplorerModel) ForEachDailyCostByDateRangeExcludingUnits(tenantID uint, billingConfigID uint, startDate, endDate time.Time, excludeUnits []string, handle func(entity.CostsDetailsQuery) error) error {
	query, params, queryTimeout := buildDailyCostsByDateRangeQuery(tenantID, billingConfigID, startDate, endDate, excludeUnits)

	days := splitDateRangeByDay(startDate, endDate)
	if len(days) > 1 {
		splitCtx, splitCancel := context.WithTimeout(context.Background(),
			getSplitLoopTimeout(false, time.Time{}))
		defer splitCancel()

		for _, day := range days {
			if err := splitCtx.Err(); err != nil {
				return billing_errors.NewSystemError(fmt.Errorf("daily cost query cancelled: %w", err))
			}
			dayParams := copyCostQueryParams(params)
			dayParams["start_date"] = day.start.Unix() - 1
			dayParams["end_date"] = day.end.Unix()
			if err := streamCostsDetailRows(query, dayParams, queryTimeout, false, []uint{tenantID}, handle); err != nil {
				return err
			}
		}
		return nil
	}
	return streamCostsDetailRows(query, params, queryTimeout, false, []uint{tenantID}, handle)
}

// ForEachDailyCostByDateRange streams the daily aggregated costs row-by-row.
func (c *CostExplorerModel) ForEachDailyCostByDateRange(tenantID uint, billingConfigID uint, startDate, endDate time.Time, handle func(entity.CostsDetailsQuery) error) error {
	return c.ForEachDailyCostByDateRangeExcludingUnits(tenantID, billingConfigID, startDate, endDate, nil, handle)
}

// buildDateRangeDetailQuery builds a daily aggregated detail query for a single tenant with an
// optional set of extra WHERE predicates (e.g. a unit REGEXP/LIKE or cluster_id IN filter). It
// reproduces the legacy getCostsSliceForDateRange* queries exactly.
func buildDateRangeDetailQuery(tenantID uint, billingConfigID uint, startDate, endDate time.Time, extraFilters []string, extraParams map[string]interface{}) (string, map[string]interface{}, time.Duration) {
	filter := []string{
		"tenant_id = @tenant_id",
		"status = @status",
		"cost_time_stamp > @start_date",
		"cost_time_stamp < @end_date",
		"internal = @internal",
	}
	params := map[string]interface{}{
		"tenant_id":  tenantID,
		"status":     entity.UserUnitCostStatus_DONE,
		"start_date": startDate.Unix() - 1,
		"end_date":   endDate.Unix(),
		"internal":   false,
	}
	filter = append(filter, extraFilters...)
	for k, v := range extraParams {
		params[k] = v
	}
	if billingConfigID != 0 {
		filter = append(filter, "binded_billing_config_id = @binded_billing_config_id")
		params["binded_billing_config_id"] = billingConfigID
	}

	query := fmt.Sprintf(`select tenant_id, binded_billing_config_id, substr(from_unixtime(cost_time_stamp), 1, 10) as billed_date, unit, labels, consumed_source, sum(cost_price) as cost_price, sum(raw_price) as raw_price
    from user_unit_costs %s where %s
    group by tenant_id, binded_billing_config_id, unit, labels, consumed_source, substr(from_unixtime(cost_time_stamp), 1, 10)
    order by substr(from_unixtime(cost_time_stamp), 1, 10);`, forceClusterIndex, strings.Join(filter, " and "))

	queryTimeout := config.Get().GetDuration(config.PerDBQueryTimeoutSeconds) * time.Second
	return query, params, queryTimeout
}

// forEachCostsDetailForDateRange streams the plain daily detail aggregation for one tenant/day.
func forEachCostsDetailForDateRange(tenantID uint, billingConfigID uint, startDate, endDate time.Time, handle func(entity.CostsDetailsQuery) error) error {
	return forEachDetailWithSnapshotOverlay(tenantID, billingConfigID, startDate, endDate, 0,
		nil,
		func(start, end time.Time, handle func(entity.CostsDetailsQuery) error) error {
			return forEachLegacyCostsDetailForDateRange(tenantID, billingConfigID, start, end, handle)
		}, handle)
}

func forEachCostsDetailForDateRangeWithCluster(tenantID uint, billingConfigID uint, startDate, endDate time.Time, clusterID uint, handle func(entity.CostsDetailsQuery) error) error {
	return forEachDetailWithSnapshotOverlay(tenantID, billingConfigID, startDate, endDate, 0,
		func(fact entity.BillingSnapshotSettlementFact) bool {
			return fact.Labels[constant.ClusterLabelKey] == fmt.Sprint(clusterID)
		},
		func(start, end time.Time, handle func(entity.CostsDetailsQuery) error) error {
			query, params, queryTimeout := buildDateRangeDetailQuery(tenantID, billingConfigID, start, end,
				[]string{"cluster_id = @cluster_id"}, map[string]interface{}{"cluster_id": clusterID})
			return streamCostsDetailRows(query, params, queryTimeout, false, []uint{tenantID}, handle)
		}, handle)
}

func forEachLegacyCostsDetailForDateRange(tenantID uint, billingConfigID uint, startDate, endDate time.Time, handle func(entity.CostsDetailsQuery) error) error {
	query, params, queryTimeout := buildDateRangeDetailQuery(tenantID, billingConfigID, startDate, endDate, nil, nil)
	return streamCostsDetailRows(query, params, queryTimeout, false, []uint{tenantID}, handle)
}

// forEachCostsDetailForDateRangeByClusterBatch streams the starter detail aggregation restricted
// to a batch of cluster IDs.
func forEachCostsDetailForDateRangeByClusterBatch(tenantID uint, billingConfigID uint, startDate, endDate time.Time, clusterIDs []string, handle func(entity.CostsDetailsQuery) error) error {
	query, params, queryTimeout := buildDateRangeDetailQuery(tenantID, billingConfigID, startDate, endDate,
		[]string{"unit REGEXP 'serverless-.*:starter$'", "cluster_id in @cluster_ids"},
		map[string]interface{}{"cluster_ids": clusterIDs})
	return streamCostsDetailRows(query, params, queryTimeout, false, []uint{tenantID}, handle)
}

// forEachCostsDetailForDateRangeByInstanceCapacity streams the instance-capacity detail
// aggregation for one tenant/day.
func forEachCostsDetailForDateRangeByInstanceCapacity(tenantID uint, billingConfigID uint, startDate, endDate time.Time, handle func(entity.CostsDetailsQuery) error) error {
	query, params, queryTimeout := buildDateRangeDetailQuery(tenantID, billingConfigID, startDate, endDate,
		[]string{"unit LIKE 'instance-capacity:%'"}, nil)
	return streamCostsDetailRows(query, params, queryTimeout, false, []uint{tenantID}, handle)
}

// s3JSONLRecordWriter is the streaming form of processCostsToRecords: it de-duplicates rows into
// S3BillsDetailRecords and flushes them to the JSONL buffer every 10k records, so the starter
// export never holds a full day's rows in memory at once. A row whose service cannot be resolved
// is skipped (matching the legacy "continue" behavior).
type s3JSONLRecordWriter struct {
	unitsCategory   map[string]string
	categoryLabels  map[string]entity.CategoryItemShow
	defaultCategory string
	uniqueKeyMap    map[string]int
	records         []forms.S3BillsDetailRecord
	buf             *bytes.Buffer
}

func newS3JSONLRecordWriter(units []forms.UnitInfo, buf *bytes.Buffer) *s3JSONLRecordWriter {
	unitsCategory := map[string]string{}
	for _, unit := range units {
		unitsCategory[unit.UnitKey] = unit.Category
	}
	return &s3JSONLRecordWriter{
		unitsCategory:   unitsCategory,
		categoryLabels:  GetAllCategoryItemInfoWithCache(),
		defaultCategory: config.Get().GetString(config.DefaultUnitCategory),
		uniqueKeyMap:    map[string]int{},
		buf:             buf,
	}
}

func (w *s3JSONLRecordWriter) add(cost entity.CostsDetailsQuery) {
	clusterIDStr := cost.Labels[constant.ClusterLabelKey]
	category := w.unitsCategory[cost.Unit]
	if category == "" {
		category = w.defaultCategory
	}
	service, ok := getServiceNameWithCategoryLabels(cost, category, w.categoryLabels)
	if !ok {
		return
	}
	var totalCost, credits decimal.Decimal
	if cost.ConsumedSource == entity.ConsumedSource_Credits {
		credits = cost.CostPrice
	} else {
		totalCost = cost.CostPrice
	}
	uniqueKey := cost.BilledDate + clusterIDStr + service + cost.ConsumedSource

	index, ok := w.uniqueKeyMap[uniqueKey]
	if ok {
		w.records[index].RunningTotal = w.records[index].RunningTotal.Add(cost.RawPrice)
		w.records[index].TotalCost = w.records[index].TotalCost.Add(totalCost)
		w.records[index].Credits = w.records[index].Credits.Add(credits)
	} else {
		w.records = append(w.records, forms.S3BillsDetailRecord{
			ClusterID:       clusterIDStr,
			BilledDate:      cost.BilledDate,
			Credits:         credits,
			RunningTotal:    cost.RawPrice,
			ServicePathName: service,
			TotalCost:       totalCost,
		})
		w.uniqueKeyMap[uniqueKey] = len(w.records) - 1
	}

	if len(w.records) >= 10000 {
		flushRecordsToJSONL(w.buf, w.records)
		w.records = w.records[:0]
		w.uniqueKeyMap = make(map[string]int)
	}
}

func (w *s3JSONLRecordWriter) flush() {
	if len(w.records) > 0 {
		flushRecordsToJSONL(w.buf, w.records)
		w.records = w.records[:0]
	}
}

// ---------------------------------------------------------------------------
// Row-by-row accumulators. Each mirrors the legacy *FromUnitCosts dedup logic so the streaming
// Get* methods and the slice-based wrappers produce identical output.
// ---------------------------------------------------------------------------

type billsDetailsAccumulator struct {
	unitsCategory   map[string]string
	categoryLabels  map[string]entity.CategoryItemShow
	defaultCategory string
	precision       int32
	uniqueKeyMap    map[string]int
	slice           []forms.UsageDetails
}

func newBillsDetailsAccumulator(units []forms.UnitInfo) *billsDetailsAccumulator {
	unitsCategory := map[string]string{}
	for _, unit := range units {
		unitsCategory[unit.UnitKey] = unit.Category
	}
	return &billsDetailsAccumulator{
		unitsCategory:   unitsCategory,
		categoryLabels:  GetAllCategoryItemInfoWithCache(),
		defaultCategory: config.Get().GetString(config.DefaultUnitCategory),
		precision:       config.Get().GetInt32(config.BillingCostsPrecisionDigits),
		uniqueKeyMap:    map[string]int{},
	}
}

// add returns false when GetServiceName fails, matching the legacy "return nil, nil" early exit.
func (a *billsDetailsAccumulator) add(cost entity.CostsDetailsQuery) bool {
	projectName := cost.Labels["project"]
	clusterName := cost.Labels["cluster"]
	category := a.unitsCategory[cost.Unit]
	if _, is := cost.Labels[constant.ServicePlanLabelKey]; is {
		projectName = ""
	}
	if category == "" {
		category = a.defaultCategory
	}
	service, ok := getServiceNameWithCategoryLabels(cost, category, a.categoryLabels)
	if !ok {
		return false
	}
	if service == "Support Plan" {
		projectName = "All Projects"
		clusterName = "All Clusters"
	}
	var billableCost, creditUsage decimal.Decimal
	if cost.ConsumedSource == entity.ConsumedSource_Credits {
		creditUsage = cost.CostPrice
	} else {
		billableCost = cost.CostPrice
	}
	uniqueKey := cost.BilledDate + projectName + clusterName + service + cost.ConsumedSource

	index, ok := a.uniqueKeyMap[uniqueKey]
	if ok {
		a.slice[index-1].Amount = a.slice[index-1].Amount.Add(cost.RawPrice)
		a.slice[index-1].BillableCost = a.slice[index-1].BillableCost.Add(billableCost)
		a.slice[index-1].CreditUsage = a.slice[index-1].CreditUsage.Add(creditUsage)
	} else {
		a.slice = append(a.slice, forms.UsageDetails{
			BilledDate:   cost.BilledDate,
			ProjectName:  projectName,
			ClusterName:  clusterName,
			Service:      service,
			BillableCost: billableCost,
			CreditUsage:  creditUsage,
			Amount:       cost.RawPrice,
		})
		a.uniqueKeyMap[uniqueKey] = len(a.slice)
	}
	return true
}

func (a *billsDetailsAccumulator) result() *forms.CostExplorerDetailsResult {
	for index, usageDetail := range a.slice {
		amount, creditUsage, billableCost := usageDetail.Amount.Truncate(a.precision), usageDetail.CreditUsage.Truncate(a.precision), usageDetail.BillableCost.Truncate(a.precision)
		a.slice[index].DiscountCost = amount.Sub(creditUsage).Sub(billableCost)
		a.slice[index].Amount = amount
		a.slice[index].BillableCost = billableCost
		a.slice[index].CreditUsage = creditUsage
	}
	return &forms.CostExplorerDetailsResult{Details: a.slice}
}

type s3BillsDetailsAccumulator struct {
	unitsCategory   map[string]string
	categoryLabels  map[string]entity.CategoryItemShow
	defaultCategory string
	precision       int32
	uniqueKeyMap    map[string]int
	records         []forms.S3BillsDetailRecord
}

func newS3BillsDetailsAccumulator(units []forms.UnitInfo) *s3BillsDetailsAccumulator {
	unitsCategory := map[string]string{}
	for _, unit := range units {
		unitsCategory[unit.UnitKey] = unit.Category
	}
	return &s3BillsDetailsAccumulator{
		unitsCategory:   unitsCategory,
		categoryLabels:  GetAllCategoryItemInfoWithCache(),
		defaultCategory: config.Get().GetString(config.DefaultUnitCategory),
		precision:       config.Get().GetInt32(config.BillingCostsPrecisionDigits),
		uniqueKeyMap:    map[string]int{},
	}
}

func (a *s3BillsDetailsAccumulator) add(cost entity.CostsDetailsQuery) bool {
	clusterIDStr := cost.Labels[constant.ClusterLabelKey]
	category := a.unitsCategory[cost.Unit]
	if category == "" {
		category = a.defaultCategory
	}
	service, ok := getServiceNameWithCategoryLabels(cost, category, a.categoryLabels)
	if !ok {
		return false
	}
	var totalCost, credits decimal.Decimal
	if cost.ConsumedSource == entity.ConsumedSource_Credits {
		credits = cost.CostPrice
	} else {
		totalCost = cost.CostPrice
	}
	uniqueKey := cost.BilledDate + clusterIDStr + service + cost.ConsumedSource

	index, ok := a.uniqueKeyMap[uniqueKey]
	if ok {
		a.records[index].RunningTotal = a.records[index].RunningTotal.Add(cost.RawPrice)
		a.records[index].TotalCost = a.records[index].TotalCost.Add(totalCost)
		a.records[index].Credits = a.records[index].Credits.Add(credits)
	} else {
		a.records = append(a.records, forms.S3BillsDetailRecord{
			ClusterID:       clusterIDStr,
			BilledDate:      cost.BilledDate,
			Credits:         credits,
			RunningTotal:    cost.RawPrice,
			ServicePathName: service,
			TotalCost:       totalCost,
		})
		a.uniqueKeyMap[uniqueKey] = len(a.records) - 1
	}
	return true
}

func (a *s3BillsDetailsAccumulator) result() *forms.S3BillsDetailResult {
	for i := range a.records {
		runningTotal := a.records[i].RunningTotal.Truncate(a.precision)
		credits := a.records[i].Credits.Truncate(a.precision)
		totalCost := a.records[i].TotalCost.Truncate(a.precision)
		a.records[i].RunningTotal = runningTotal
		a.records[i].Credits = credits
		a.records[i].TotalCost = totalCost
		a.records[i].Discounts = runningTotal.Sub(credits).Sub(totalCost)
	}
	return &forms.S3BillsDetailResult{Details: a.records}
}

type partnerBillsDetailsAccumulator struct {
	unitsCategory      map[string]string
	categoryLabels     map[string]entity.CategoryItemShow
	defaultCategory    string
	precision          int32
	tenantIDAsCustomer bool
	uniqueKeyMap       map[string]int
	slice              []forms.PartnerBillsDetail
}

func newPartnerBillsDetailsAccumulator(units []forms.UnitInfo, tenantIDAsCustomer bool) *partnerBillsDetailsAccumulator {
	unitsCategory := map[string]string{}
	for _, unit := range units {
		unitsCategory[unit.UnitKey] = unit.Category
	}
	return &partnerBillsDetailsAccumulator{
		unitsCategory:      unitsCategory,
		categoryLabels:     GetAllCategoryItemInfoWithCache(),
		defaultCategory:    config.Get().GetString(config.DefaultUnitCategory),
		precision:          config.Get().GetInt32(config.BillingCostsPrecisionDigits),
		tenantIDAsCustomer: tenantIDAsCustomer,
		uniqueKeyMap:       map[string]int{},
	}
}

// add returns false when GetServiceName fails, matching the legacy nil-result behavior.
func (a *partnerBillsDetailsAccumulator) add(cost entity.CostsDetailsQuery) (bool, error) {
	projectID := cost.Labels[constant.ProjectLabelKey]
	projectName := cost.Labels[constant.ProjectNameLabelKey]
	clusterID := cost.Labels[constant.ClusterLabelKey]
	clusterName := cost.Labels[constant.ClusterNameLabelKey]
	customerIDStr := cost.Labels[constant.MSPCustomerIDLabel]
	if a.tenantIDAsCustomer {
		customerIDStr = fmt.Sprint(cost.TenantID)
	}
	category := a.unitsCategory[cost.Unit]
	if category == "" {
		category = a.defaultCategory
	}
	service, ok := getServiceNameWithCategoryLabels(cost, category, a.categoryLabels)
	if !ok {
		return false, nil
	}
	if service == "Support Plan" {
		projectName = "All Projects"
		clusterName = "All Clusters"
	}
	var billableCost, creditUsage decimal.Decimal
	if cost.ConsumedSource == entity.ConsumedSource_Credits {
		creditUsage = cost.CostPrice
	} else {
		billableCost = cost.CostPrice
	}
	uniqueKey := cost.BilledDate + projectName + clusterName + service + cost.ConsumedSource + customerIDStr

	index, ok := a.uniqueKeyMap[uniqueKey]
	if ok {
		a.slice[index-1].RunningTotal = a.slice[index-1].RunningTotal.Add(cost.RawPrice)
		a.slice[index-1].TotalCost = a.slice[index-1].TotalCost.Add(billableCost)
		a.slice[index-1].Credits = a.slice[index-1].Credits.Add(creditUsage)
	} else {
		a.slice = append(a.slice, forms.PartnerBillsDetail{
			BillsDetail: forms.BillsDetail{
				BillsCosts: forms.BillsCosts{
					RunningTotal: cost.RawPrice,
					Credits:      creditUsage,
					TotalCost:    billableCost,
				},
				BilledDate:      cost.BilledDate,
				ProjectID:       projectID,
				ProjectName:     projectName,
				ClusterID:       clusterID,
				ClusterName:     clusterName,
				ServicePathName: service,
			},
			CustomerID: customerIDStr,
		})
		a.uniqueKeyMap[uniqueKey] = len(a.slice)
	}
	return true, nil
}

func (a *partnerBillsDetailsAccumulator) result() ([]forms.PartnerBillsDetail, error) {
	customerNames := map[string]string{}
	for index, usageDetail := range a.slice {
		if usageDetail.CustomerID != "" {
			customerName, ok := customerNames[usageDetail.CustomerID]
			if !ok {
				parsed, err := strconv.ParseUint(usageDetail.CustomerID, 10, 64)
				if err != nil {
					return nil, err
				}
				customerID := uint(parsed)
				org, err := centralhandler.GetCentralHandler().GetTenant(customerID)
				if err != nil {
					return nil, err
				}
				customerName = org.Name
				customerNames[usageDetail.CustomerID] = customerName
			}
			a.slice[index].CustomerName = customerName
		}
		amount, creditUsage, billableCost := usageDetail.RunningTotal.Truncate(a.precision), usageDetail.Credits.Truncate(a.precision), usageDetail.TotalCost.Truncate(a.precision)
		a.slice[index].Discounts = amount.Sub(creditUsage).Sub(billableCost)
		a.slice[index].RunningTotalTruncatedLoss = a.slice[index].RunningTotal.Sub(amount)
		a.slice[index].RunningTotal = amount
		a.slice[index].TotalCostTruncatedLoss = a.slice[index].TotalCost.Sub(billableCost)
		a.slice[index].TotalCost = billableCost
		a.slice[index].CreditsTruncatedLoss = a.slice[index].Credits.Sub(creditUsage)
		a.slice[index].Credits = creditUsage
	}
	return a.slice, nil
}

type billsExportDetailsAccumulator struct {
	unitsCategory  map[string]string
	unitsProvider  map[string]string
	unitsRegion    map[string]string
	unitsPrice     map[string]decimal.Decimal
	categoryLabels map[string]entity.CategoryItemShow
	usageContext   exportUsageContext
	uniqueKeyMap   map[string]int
	slice          []forms.ExportUsageDetails
}

func newBillsExportDetailsAccumulator(units []forms.UnitInfo, usageContext exportUsageContext) *billsExportDetailsAccumulator {
	unitsCategory := map[string]string{}
	unitsProvider := map[string]string{}
	unitsRegion := map[string]string{}
	unitsPrice := map[string]decimal.Decimal{}
	for _, unit := range units {
		unitsCategory[unit.UnitKey] = unit.Category
		unitsPrice[unit.UnitKey] = unit.Price
		if len(unit.Provider) > 0 {
			unitsProvider[unit.UnitKey] = unit.Provider
		} else {
			unitsProvider[unit.UnitKey] = "-"
		}
		if len(unit.Region) > 0 {
			unitsRegion[unit.UnitKey] = unit.Region
		} else {
			unitsRegion[unit.UnitKey] = "-"
		}
	}
	return &billsExportDetailsAccumulator{
		unitsCategory:  unitsCategory,
		unitsProvider:  unitsProvider,
		unitsRegion:    unitsRegion,
		unitsPrice:     unitsPrice,
		categoryLabels: GetAllCategoryItemInfoWithCache(),
		usageContext:   usageContext,
		uniqueKeyMap:   map[string]int{},
	}
}

func (a *billsExportDetailsAccumulator) add(cost entity.CostsDetailsQuery) bool {
	cloudProvider := a.unitsProvider[cost.Unit]
	region := a.unitsRegion[cost.Unit]
	projectName := cost.Labels["project"]
	projectID := cost.Labels["projectID"]
	clusterName := cost.Labels["cluster"]
	clusterID := cost.Labels["clusterID"]
	if _, is := cost.Labels[constant.ServicePlanLabelKey]; is {
		projectName = ""
	}
	category := a.unitsCategory[cost.Unit]
	categoryItemInfo := a.categoryLabels[category]
	service, ok := getServiceNameWithCategoryLabels(cost, category, a.categoryLabels)
	if !ok {
		return false
	}
	if service == "Support Plan" {
		projectName = "All Projects"
		clusterName = "All Clusters"
	}
	usage, usageUnit := getExportUsage(cost, categoryItemInfo, a.unitsPrice[cost.Unit], a.usageContext)
	var billableCost, creditUsage decimal.Decimal
	if cost.ConsumedSource == entity.ConsumedSource_Credits {
		creditUsage = cost.CostPrice
	} else {
		billableCost = cost.CostPrice
	}
	uniqueKey := fmt.Sprint(cost.TenantID) + cost.BilledDate + cloudProvider + region + projectName + projectID + clusterName + clusterID + service + cost.ConsumedSource
	index, ok := a.uniqueKeyMap[uniqueKey]
	if ok {
		a.slice[index-1].Amount = a.slice[index-1].Amount.Add(cost.RawPrice)
		a.slice[index-1].BillableCost = a.slice[index-1].BillableCost.Add(billableCost)
		a.slice[index-1].CreditUsage = a.slice[index-1].CreditUsage.Add(creditUsage)
		mergeExportUsage(&a.slice[index-1], usage, usageUnit)
	} else {
		a.slice = append(a.slice, forms.ExportUsageDetails{
			TenantID:        cost.TenantID,
			BillingConfigID: cost.BindedBillingConfigID,
			BilledDate:      cost.BilledDate,
			Service:         service,
			Usage:           usage,
			Unit:            usageUnit,
			CloudProvider:   cloudProvider,
			Region:          region,
			ProjectName:     projectName,
			ProjectID:       projectID,
			ClusterName:     clusterName,
			ClusterID:       clusterID,
			Amount:          cost.RawPrice,
			BillableCost:    billableCost,
			CreditUsage:     creditUsage,
		})
		a.uniqueKeyMap[uniqueKey] = len(a.slice)
	}
	return true
}

func (a *billsExportDetailsAccumulator) result() []forms.ExportUsageDetails {
	return a.slice
}