package tests

import (
	"crypto/md5"
	"encoding/json"
	"fmt"
	"sort"
	"strconv"
	"time"

	"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/handler/paymenthandler"
	"github.com/pingcap/dbaas/billing2/handler/paymenthandler/impl"
	"github.com/pingcap/dbaas/billing2/model/billing"
	"github.com/pingcap/dbaas/billing2/model/billingconfig"
	"github.com/pingcap/dbaas/billing2/utils/tests/datainit"
	"github.com/pingcap/dbaas/billing2/utils/timeutils"
	"github.com/shopspring/decimal"
)

// transferHash re-derives the deterministic transfer costID hash independently
// of impl.getHashString/buildTransferMeasure (which live in the impl package
// and are unexported): md5 over sorted key+value pairs, then unit, cost,
// dateUnix, then any extra domain strings. Used to assert the Pay costIDs
// without calling the code under test.
func transferHash(unit string, cost int, dateUnix int64, labels map[string]string, extra ...string) string {
	keys := make([]string, 0, len(labels))
	for key := range labels {
		keys = append(keys, key)
	}
	sort.Strings(keys)

	hasher := md5.New()
	for _, key := range keys {
		_, _ = hasher.Write([]byte(key + labels[key]))
	}
	_, _ = hasher.Write([]byte(unit))
	_, _ = hasher.Write([]byte(strconv.Itoa(cost)))
	_, _ = hasher.Write([]byte(strconv.FormatInt(dateUnix, 10)))
	for _, e := range extra {
		_, _ = hasher.Write([]byte(e))
	}
	return fmt.Sprintf("%x", hasher.Sum(nil))
}

// expectedLegacyTransferCostID computes the pre-replay single-row transfer
// costID domain: day-granularity hash over the row labels plus the customer
// attribution label, no extra domain string.
func expectedLegacyTransferCostID(invoiceID uint, customerIDLabel, customerIDStr, unit string, cost int, date time.Time, labels map[string]string) string {
	merged := make(map[string]string, len(labels)+1)
	for k, v := range labels {
		merged[k] = v
	}
	merged[customerIDLabel] = customerIDStr
	return fmt.Sprintf("%d-%s", invoiceID, transferHash(unit, cost, date.Unix(), merged))
}

// expectedReplayTransferCostID computes the replay costID domain: full
// per-entry timestamp plus the source raw costID as the extra domain string.
func expectedReplayTransferCostID(invoiceID uint, customerIDLabel, customerIDStr, unit string, cost int, costTimeStamp int64, labels map[string]string, sourceCostID string) string {
	merged := make(map[string]string, len(labels)+1)
	for k, v := range labels {
		merged[k] = v
	}
	merged[customerIDLabel] = customerIDStr
	return fmt.Sprintf("%d-%s", invoiceID, transferHash(unit, cost, costTimeStamp, merged, sourceCostID))
}

// mustMarshalHistory serializes a MergedLabelsHistory for the raw SQL
// history fix-ups in the partial-failure phase.
func mustMarshalHistory(history entity.MergedLabelsHistory) string {
	data, err := json.Marshal(history)
	if err != nil {
		panic(err)
	}
	return string(data)
}

// TestMSPPayReplaysMergedLabelsHistory is the end-to-end test for
// MspHandlerImpl.Pay's merged-labels replay:
//
//  1. A merged customer row with a complete two-entry history must be
//     replayed to the partner as TWO SecondaryMeasure transfers, each
//     carrying its own entry labels and deterministic replay costID.
//  2. Re-running Pay (invoice retry) must not double-transfer: the same
//     replay costIDs dedup on the partner side and no new rows appear.
//  3. Partial failure: when a third entry fails to transfer mid-Pay (its
//     billing-config resolution fails), the already-transferred entries
//     stay on the partner books and the Pay returns an error; after the
//     entry is repaired, a retried Pay transfers only the missing entry.
func (suite *PaymentSuite) TestMSPPayReplaysMergedLabelsHistory() {
	assert := suite.Assert()
	centralhandler.SetCentralHandler(&MockPartnerCentralHandlerImpl{})
	paymenthandler.SetPaymentHandler(&impl.PaymentHandlerImpl{})
	timeutils.SetMockNow(datainit.BillingDate)

	const customerTenantID uint = 40
	const resellerTenantID uint = 50
	const testInvoiceID = 99002

	_, err := datainit.CreateBillingConfig(customerTenantID, entity.BillingConfigType_DirectCustomer)
	assert.NoError(err)
	_, err = datainit.CreateBillingConfig(resellerTenantID, entity.BillingConfigType_DirectCustomer)
	assert.NoError(err)
	billingConfigModel := &billingconfig.BillingConfigModel{}
	customerBC, err := billingConfigModel.GetActiveBillingConfig(customerTenantID)
	assert.NoError(err)
	resellerBC, err := billingConfigModel.GetActiveBillingConfig(resellerTenantID)
	assert.NoError(err)

	_, err = (&billing.UnitModel{}).CreateUnit(&forms.UnitInfo{
		UnitKey:   datainit.TestUnit,
		Price:     decimal.NewFromInt(1),
		PriceType: entity.PriceType_Normal,
	})
	assert.NoError(err)

	invoiceDate := datainit.BillingDate.AddDate(0, 0, 5)
	ts1 := invoiceDate.Add(time.Hour).Unix()
	ts2 := invoiceDate.Add(2 * time.Hour).Unix()

	mergedRow := &entity.UserUnitCost{
		CostID:                "test-pay-replay-merged",
		TenantID:              customerTenantID,
		CostTimeStamp:         ts2,
		Unit:                  datainit.TestUnit,
		Cost:                  150,
		CostPrice:             decimal.NewFromInt(150),
		RawPrice:              decimal.NewFromInt(150),
		Status:                entity.UserUnitCostStatus_DONE,
		BindedBillingConfigID: customerBC.BillingConfigID,
		InvoiceID:             testInvoiceID,
		ConsumedSource:        entity.ConsumedSource_Invoice,
		// First-chunk labels: without replay, these stale labels would price
		// BOTH contributions with intervalValue=1.
		Labels: entity.Labels{"clusterID": "c1", "intervalValue": "1"},
		MergedLabelsHistory: entity.MergedLabelsHistory{
			{CostID: "test-pay-replay-raw-1", Cost: 100, CostTimeStamp: ts1, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "1"}},
			{CostID: "test-pay-replay-raw-2", Cost: 50, CostTimeStamp: ts2, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "2"}},
		},
	}
	assert.NoError(mergedRow.Create(db.GetDB()))

	invoiceEntity := &entity.Invoice{
		TenantID:              customerTenantID,
		Status:                entity.InvoiceStatus_SENT,
		Type:                  entity.InvoiceType_MONTHLY,
		InvoiceDate:           invoiceDate,
		BindedBillingConfigID: customerBC.BillingConfigID,
		CostPrice:             decimal.NewFromInt(150),
		RawPrice:              decimal.NewFromInt(150),
	}
	invoiceEntity.ID = testInvoiceID
	assert.NoError(db.GetDB().Create(invoiceEntity).Error)

	paymentInfo := &forms.PaymentMethod{TenantID: customerTenantID, PaymentCode: fmt.Sprint(resellerTenantID)}
	formInvoice := &forms.Invoice{
		InvoiceID: testInvoiceID,
		TenantID:  customerTenantID,
		Status:    entity.InvoiceStatus_SENT,
		Type:      entity.InvoiceType_MONTHLY,
		StartDate: invoiceDate,
		EndDate:   invoiceDate,
	}

	queryPartnerRows := func() []entity.UserUnitCost {
		var rows []entity.UserUnitCost
		assert.NoError(entity.NewUserUnitCostQuerySet(db.GetDB()).
			TenantIDEq(resellerTenantID).
			BindedBillingConfigIDEq(resellerBC.BillingConfigID).
			UnitEq(datainit.TestUnit).
			All(&rows))
		sort.Slice(rows, func(i, j int) bool { return rows[i].Cost < rows[j].Cost })
		return rows
	}
	queryPartnerRawCostIDs := func() []string {
		var raws []entity.RawUserUnitCost
		assert.NoError(entity.NewRawUserUnitCostQuerySet(db.GetDB()).
			TenantIDEq(resellerTenantID).
			All(&raws))
		costIDs := make([]string, 0, len(raws))
		for _, r := range raws {
			costIDs = append(costIDs, r.CostID)
		}
		sort.Strings(costIDs)
		return costIDs
	}

	// --- Phase A: complete two-entry history must replay per-entry. ---
	_, err = (&impl.MspHandlerImpl{}).Pay(paymentInfo, nil, formInvoice)
	assert.NoError(err)

	partnerRows := queryPartnerRows()
	assert.Len(partnerRows, 2, "expected two per-entry transfers on the partner books")
	assert.Equal(50, partnerRows[0].Cost)
	assert.Equal(100, partnerRows[1].Cost)
	assert.Equal(entity.Labels{"clusterID": "c1", "intervalValue": "2", constant.MSPCustomerIDLabel: "40"}, partnerRows[0].Labels)
	assert.Equal(entity.Labels{"clusterID": "c1", "intervalValue": "1", constant.MSPCustomerIDLabel: "40"}, partnerRows[1].Labels)

	rawCostIDs := queryPartnerRawCostIDs()
	assert.Len(rawCostIDs, 2, "expected two partner raw rows, one per replayed entry")
	expectedIDs := []string{
		expectedReplayTransferCostID(testInvoiceID, constant.MSPCustomerIDLabel, "40", datainit.TestUnit, 50, ts2, entity.Labels{"clusterID": "c1", "intervalValue": "2"}, "test-pay-replay-raw-2"),
		expectedReplayTransferCostID(testInvoiceID, constant.MSPCustomerIDLabel, "40", datainit.TestUnit, 100, ts1, entity.Labels{"clusterID": "c1", "intervalValue": "1"}, "test-pay-replay-raw-1"),
	}
	sort.Strings(expectedIDs)
	assert.Equal(expectedIDs, rawCostIDs)

	// --- Phase A retry: re-Pay must not double-transfer. ---
	_, err = (&impl.MspHandlerImpl{}).Pay(paymentInfo, nil, formInvoice)
	assert.NoError(err)
	partnerRows = queryPartnerRows()
	assert.Len(partnerRows, 2, "retried Pay must not create extra partner rows")
	assert.Equal(50, partnerRows[0].Cost)
	assert.Equal(100, partnerRows[1].Cost)
	assert.Equal(rawCostIDs, queryPartnerRawCostIDs(), "retried Pay must reuse the same deterministic replay costIDs")

	// --- Phase B: partial failure — a third entry whose billing-config
	// resolution fails (CostTimeStamp=0) aborts the Pay after the first two
	// entries committed. ---
	historyWithBrokenEntry := entity.MergedLabelsHistory{
		{CostID: "test-pay-replay-raw-1", Cost: 100, CostTimeStamp: ts1, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "1"}},
		{CostID: "test-pay-replay-raw-2", Cost: 50, CostTimeStamp: ts2, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "2"}},
		{CostID: "test-pay-replay-raw-3", Cost: 25, CostTimeStamp: 0, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "3"}},
	}
	assert.NoError(db.GetDB().Exec(
		"UPDATE user_unit_costs SET cost = ?, merged_labels_history = ? WHERE cost_id = ?",
		175, mustMarshalHistory(historyWithBrokenEntry), "test-pay-replay-merged").Error)

	_, err = (&impl.MspHandlerImpl{}).Pay(paymentInfo, nil, formInvoice)
	assert.Error(err, "Pay must fail when an entry cannot resolve its billing config")
	partnerRows = queryPartnerRows()
	assert.Len(partnerRows, 2, "the failed entry must not leave partner rows behind")
	assert.Equal(rawCostIDs, queryPartnerRawCostIDs(), "the failed entry must not leave a partner raw row behind")

	// --- Phase B retry: repair the entry's timestamp; the retried Pay
	// transfers only the missing entry — the two already-transferred
	// entries dedup via their deterministic costIDs. ---
	ts3 := invoiceDate.Add(3 * time.Hour).Unix()
	historyFixed := entity.MergedLabelsHistory{
		{CostID: "test-pay-replay-raw-1", Cost: 100, CostTimeStamp: ts1, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "1"}},
		{CostID: "test-pay-replay-raw-2", Cost: 50, CostTimeStamp: ts2, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "2"}},
		{CostID: "test-pay-replay-raw-3", Cost: 25, CostTimeStamp: ts3, Labels: entity.Labels{"clusterID": "c1", "intervalValue": "3"}},
	}
	assert.NoError(db.GetDB().Exec(
		"UPDATE user_unit_costs SET cost = ?, merged_labels_history = ? WHERE cost_id = ?",
		175, mustMarshalHistory(historyFixed), "test-pay-replay-merged").Error)

	_, err = (&impl.MspHandlerImpl{}).Pay(paymentInfo, nil, formInvoice)
	assert.NoError(err)
	partnerRows = queryPartnerRows()
	assert.Len(partnerRows, 3, "repaired retry must transfer exactly the missing entry")
	sort.Slice(partnerRows, func(i, j int) bool { return partnerRows[i].Cost < partnerRows[j].Cost })
	assert.Equal(25, partnerRows[0].Cost)
	assert.Equal(50, partnerRows[1].Cost)
	assert.Equal(100, partnerRows[2].Cost)
	assert.Equal(entity.Labels{"clusterID": "c1", "intervalValue": "3", constant.MSPCustomerIDLabel: "40"}, partnerRows[0].Labels)

	rawCostIDsAfterRepair := queryPartnerRawCostIDs()
	assert.Len(rawCostIDsAfterRepair, 3, "the repaired retry must add exactly one new raw row")
	expectedIDsAfterRepair := append(rawCostIDs,
		expectedReplayTransferCostID(testInvoiceID, constant.MSPCustomerIDLabel, "40", datainit.TestUnit, 25, ts3, entity.Labels{"clusterID": "c1", "intervalValue": "3"}, "test-pay-replay-raw-3"))
	sort.Strings(expectedIDsAfterRepair)
	assert.Equal(expectedIDsAfterRepair, rawCostIDsAfterRepair, "already-transferred entries must keep their original costIDs")

	// The customer-side row is untouched by Pay.
	var customerRow entity.UserUnitCost
	assert.NoError(entity.NewUserUnitCostQuerySet(db.GetDB()).CostIDEq("test-pay-replay-merged").One(&customerRow))
	assert.Equal(175, customerRow.Cost)
}

// TestCPPOPayTransfersMergedLabelsHistory is the end-to-end test for
// AWSCPPOCustomerInvoiceTransferHandlerImpl.Pay:
//
//  1. A row WITHOUT history must take the legacy single-row transfer whose
//     costID is byte-identical with the pre-feature transfer.
//  2. A row WITH a complete history must replay per-entry with each entry's
//     own labels and replay-domain costIDs.
//  3. Re-running Pay must not double-transfer either shape.
func (suite *PaymentSuite) TestCPPOPayTransfersMergedLabelsHistory() {
	assert := suite.Assert()
	centralhandler.SetCentralHandler(&MockPartnerCentralHandlerImpl{})
	paymenthandler.SetPaymentHandler(&impl.PaymentHandlerImpl{})
	timeutils.SetMockNow(datainit.BillingDate)

	const customerTenantID uint = 70
	const payerTenantID uint = 60
	const testInvoiceID = 99003

	_, err := datainit.CreateBillingConfig(customerTenantID, entity.BillingConfigType_DirectCustomer)
	assert.NoError(err)
	_, err = datainit.CreateBillingConfig(payerTenantID, entity.BillingConfigType_DirectCustomer)
	assert.NoError(err)
	billingConfigModel := &billingconfig.BillingConfigModel{}
	customerBC, err := billingConfigModel.GetActiveBillingConfig(customerTenantID)
	assert.NoError(err)
	payerBC, err := billingConfigModel.GetActiveBillingConfig(payerTenantID)
	assert.NoError(err)

	_, err = (&billing.UnitModel{}).CreateUnit(&forms.UnitInfo{
		UnitKey:   datainit.TestUnit,
		Price:     decimal.NewFromInt(1),
		PriceType: entity.PriceType_Normal,
	})
	assert.NoError(err)

	invoiceDate := datainit.BillingDate.AddDate(0, 0, 5)
	ts := invoiceDate.Add(3 * time.Hour).Unix()
	ts1 := invoiceDate.Add(time.Hour).Unix()
	ts2 := invoiceDate.Add(2 * time.Hour).Unix()
	customerIDStr := strconv.FormatUint(uint64(customerTenantID), 10)

	// Same unit/date/labels — only merged_labels_history separates the two
	// rows, so they must produce two transfers: one legacy fallback, one replay.
	fallbackRow := &entity.UserUnitCost{
		CostID:                "cppo-pay-fallback-row",
		TenantID:              customerTenantID,
		CostTimeStamp:         ts,
		Unit:                  datainit.TestUnit,
		Cost:                  300,
		CostPrice:             decimal.NewFromInt(300),
		RawPrice:              decimal.NewFromInt(300),
		Status:                entity.UserUnitCostStatus_DONE,
		BindedBillingConfigID: customerBC.BillingConfigID,
		InvoiceID:             testInvoiceID,
		ConsumedSource:        entity.ConsumedSource_Invoice,
		Labels:                entity.Labels{"clusterID": "c2"},
	}
	assert.NoError(fallbackRow.Create(db.GetDB()))
	replayRow := &entity.UserUnitCost{
		CostID:                "cppo-pay-replay-row",
		TenantID:              customerTenantID,
		CostTimeStamp:         ts,
		Unit:                  datainit.TestUnit,
		Cost:                  150,
		CostPrice:             decimal.NewFromInt(150),
		RawPrice:              decimal.NewFromInt(150),
		Status:                entity.UserUnitCostStatus_DONE,
		BindedBillingConfigID: customerBC.BillingConfigID,
		InvoiceID:             testInvoiceID,
		ConsumedSource:        entity.ConsumedSource_Invoice,
		Labels:                entity.Labels{"clusterID": "c2"},
		MergedLabelsHistory: entity.MergedLabelsHistory{
			{CostID: "cppo-pay-replay-raw-1", Cost: 100, CostTimeStamp: ts1, Labels: entity.Labels{"clusterID": "c2", "intervalValue": "1"}},
			{CostID: "cppo-pay-replay-raw-2", Cost: 50, CostTimeStamp: ts2, Labels: entity.Labels{"clusterID": "c2", "intervalValue": "2"}},
		},
	}
	assert.NoError(replayRow.Create(db.GetDB()))

	invoiceEntity := &entity.Invoice{
		TenantID:              customerTenantID,
		Status:                entity.InvoiceStatus_SENT,
		Type:                  entity.InvoiceType_MONTHLY,
		InvoiceDate:           invoiceDate,
		BindedBillingConfigID: customerBC.BillingConfigID,
		CostPrice:             decimal.NewFromInt(450),
		RawPrice:              decimal.NewFromInt(450),
	}
	invoiceEntity.ID = testInvoiceID
	assert.NoError(db.GetDB().Create(invoiceEntity).Error)

	paymentInfo := &forms.PaymentMethod{
		TenantID:    customerTenantID,
		PaymentCode: fmt.Sprintf("%d-%d", payerTenantID, payerBC.BillingConfigID),
	}
	formInvoice := &forms.Invoice{
		InvoiceID: testInvoiceID,
		TenantID:  customerTenantID,
		Status:    entity.InvoiceStatus_SENT,
		Type:      entity.InvoiceType_MONTHLY,
		StartDate: invoiceDate,
		EndDate:   invoiceDate,
	}

	queryPayerRows := func() []entity.UserUnitCost {
		var rows []entity.UserUnitCost
		assert.NoError(entity.NewUserUnitCostQuerySet(db.GetDB()).
			TenantIDEq(payerTenantID).
			BindedBillingConfigIDEq(payerBC.BillingConfigID).
			UnitEq(datainit.TestUnit).
			All(&rows))
		sort.Slice(rows, func(i, j int) bool { return rows[i].Cost < rows[j].Cost })
		return rows
	}
	queryPayerRawCostIDs := func() []string {
		var raws []entity.RawUserUnitCost
		assert.NoError(entity.NewRawUserUnitCostQuerySet(db.GetDB()).
			TenantIDEq(payerTenantID).
			All(&raws))
		costIDs := make([]string, 0, len(raws))
		for _, r := range raws {
			costIDs = append(costIDs, r.CostID)
		}
		sort.Strings(costIDs)
		return costIDs
	}

	_, err = (&impl.AWSCPPOCustomerInvoiceTransferHandlerImpl{}).Pay(paymentInfo, nil, formInvoice)
	assert.NoError(err)

	payerRows := queryPayerRows()
	assert.Len(payerRows, 3, "expected the fallback transfer plus two replayed entries")
	assert.Equal(50, payerRows[0].Cost)
	assert.Equal(100, payerRows[1].Cost)
	assert.Equal(300, payerRows[2].Cost)
	assert.Equal(entity.Labels{"clusterID": "c2", "intervalValue": "2", constant.MarketplaceResellerManagedCustomerIDLabel: customerIDStr}, payerRows[0].Labels)
	assert.Equal(entity.Labels{"clusterID": "c2", "intervalValue": "1", constant.MarketplaceResellerManagedCustomerIDLabel: customerIDStr}, payerRows[1].Labels)
	assert.Equal(entity.Labels{"clusterID": "c2", constant.MarketplaceResellerManagedCustomerIDLabel: customerIDStr}, payerRows[2].Labels)

	// The fallback transfer's costID must be byte-identical with the legacy
	// single-row transfer (day-granularity hash, no extra domain string).
	expectedFallbackID := expectedLegacyTransferCostID(testInvoiceID,
		constant.MarketplaceResellerManagedCustomerIDLabel, customerIDStr,
		datainit.TestUnit, 300, invoiceDate, entity.Labels{"clusterID": "c2"})
	expectedReplayIDs := []string{
		expectedReplayTransferCostID(testInvoiceID, constant.MarketplaceResellerManagedCustomerIDLabel, customerIDStr, datainit.TestUnit, 50, ts2, entity.Labels{"clusterID": "c2", "intervalValue": "2"}, "cppo-pay-replay-raw-2"),
		expectedReplayTransferCostID(testInvoiceID, constant.MarketplaceResellerManagedCustomerIDLabel, customerIDStr, datainit.TestUnit, 100, ts1, entity.Labels{"clusterID": "c2", "intervalValue": "1"}, "cppo-pay-replay-raw-1"),
	}
	expectedIDs := append(expectedReplayIDs, expectedFallbackID)
	sort.Strings(expectedIDs)
	assert.Equal(expectedIDs, queryPayerRawCostIDs())

	// Retried Pay must not double-transfer either shape.
	_, err = (&impl.AWSCPPOCustomerInvoiceTransferHandlerImpl{}).Pay(paymentInfo, nil, formInvoice)
	assert.NoError(err)
	payerRows = queryPayerRows()
	assert.Len(payerRows, 3, "retried Pay must not create extra payer rows")
	assert.Equal(expectedIDs, queryPayerRawCostIDs(), "retried Pay must reuse the same deterministic costIDs")
}