package workers

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/pingcap/dbaas/billing2/config"
	"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/model/payment"
	"github.com/pingcap/dbaas/billing2/model/payment/paymentsub"
	"github.com/pingcap/dbaas/billing2/utils/tests"
	"github.com/pingcap/dbaas/billing2/utils/tests/datainit"
	"github.com/shopspring/decimal"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
)

// MockStripePayingProvider satisfies both PaymentProvider (per-invoice
// Pay path) and StripePaymentHandler (merged PushInvoice path). The
// embedded nil interfaces cover the unused methods; only Pay /
// PayMultiInvoice / PushInvoice / Payable are called by runPaying.
type MockStripePayingProvider struct {
	mock.Mock
	paymenthandler.PaymentProvider
	paymenthandler.StripePaymentHandler

	// Records for assertions.
	PayCalls         []uint   // invoice IDs pushed via Pay
	PushInvoiceCalls [][]uint // invoice ID batches pushed via PushInvoice
}

func (m *MockStripePayingProvider) Payable(tenantID uint, userCode string) (bool, error) {
	return true, nil
}

func (*MockStripePayingProvider) Setup(tenant *forms.Tenant) (string, error) {
	return "cus_mock", nil
}

func (*MockStripePayingProvider) NeedMultiInvoiceSendOnce() bool { return false }

func (*MockStripePayingProvider) EnableHourlySending(tenantID, billingConfigID uint) bool {
	return false
}

func (*MockStripePayingProvider) EnableDailySending(tenantID, billingConfigID uint) (bool, int) {
	return false, 0
}

func (*MockStripePayingProvider) NeedInMonthInvoice() bool        { return false }
func (*MockStripePayingProvider) NeedSetInvoiceAutoConfirm() bool { return false }
func (*MockStripePayingProvider) EnableTaxCalculation() bool      { return false }
func (*MockStripePayingProvider) EnableReconciliation() bool      { return false }

// Pay mocks the per-invoice Stripe push: returns the invoice with a
// Stripe invoice ID populated. The caller (paymentsub.pay) then flips
// GENERATED → SENT and fills payment type/code.
func (m *MockStripePayingProvider) Pay(paymentInfo *forms.PaymentMethod, billingCompany *forms.BillingCompany, invoice *forms.Invoice) (*forms.Invoice, error) {
	m.PayCalls = append(m.PayCalls, invoice.InvoiceID)
	invoice.StripeInvoiceID = fmt.Sprintf("in_mock_%d", invoice.InvoiceID)
	invoice.ChargeMode = "auto"
	return invoice, nil
}

func (m *MockStripePayingProvider) PayMultiInvoice(paymentInfo *forms.PaymentMethod, billingCompany *forms.BillingCompany, invoices []forms.Invoice) ([]forms.Invoice, error) {
	for i := range invoices {
		invoices[i].StripeInvoiceID = fmt.Sprintf("in_mock_%d", invoices[i].InvoiceID)
		invoices[i].ChargeMode = "manual"
	}
	return invoices, nil
}

// PushInvoice mocks the merged push: records the batch and returns the
// first invoice with a Stripe invoice ID.
func (m *MockStripePayingProvider) PushInvoice(paymentInfo *forms.PaymentMethod, billingCompany *forms.BillingCompany, invoices []forms.Invoice, chargeMode paymenthandler.InvoiceChargeMode) (*forms.Invoice, error) {
	batch := make([]uint, 0, len(invoices))
	for _, inv := range invoices {
		batch = append(batch, inv.InvoiceID)
	}
	m.PushInvoiceCalls = append(m.PushInvoiceCalls, batch)
	out := invoices[0]
	out.StripeInvoiceID = fmt.Sprintf("in_merged_%d", invoices[0].InvoiceID)
	out.ChargeMode = string(chargeMode)
	return &out, nil
}

// MockStripePayingHandler returns the provider for any payment type.
type MockStripePayingHandler struct {
	Provider *MockStripePayingProvider
}

func (m *MockStripePayingHandler) GetProvider(string) (paymenthandler.PaymentProvider, error) {
	return m.Provider, nil
}

func (m *MockStripePayingHandler) AllProviders() []string {
	return []string{"stripe"}
}

type StripePayingWorkerSuite struct {
	tests.BillingTestSuite
	provider *MockStripePayingProvider
}

func TestStripePayingWorkerSuite(t *testing.T) {
	suite.Run(t, new(StripePayingWorkerSuite))
}

func (s *StripePayingWorkerSuite) SetupTest() {
	s.BillingTestSuite.SetupTest()
	s.provider = &MockStripePayingProvider{}
	paymenthandler.SetPaymentHandler(&MockStripePayingHandler{Provider: s.provider})
	mockCentral()

	config.Get().Set(config.StripePayingPushPerInvoice, true)
	config.Get().Set(config.StripePushMinAmountUSD, "1.00")
	config.Get().Set(config.StripePushMaxCarryoverMonths, 12)
	config.Get().Set(config.StripePushKeepDraftOnChargeModeManual, false)
}

// seedGroup creates an active stripe payment method, billing config,
// billing company, and invoices for a tenant/bc pair.
func (s *StripePayingWorkerSuite) seedGroup(tenantID, billingConfigID uint, amounts []float64, statuses []string) {
	assert := s.Assert()
	paymentModel := &paymentsub.PaymentModel{}
	_, err := paymentModel.CreatePaymentMethodAndActive(tenantID, &forms.PaymentMethodCreateQuery{
		PaymentType:           entity.PaymentType_STRIPE,
		BindedBillingConfigID: &billingConfigID,
	})
	assert.NoError(err)

	billingInfoModel := &paymentsub.BillingInformationModel{}
	assert.NoError(billingInfoModel.CreateBillingCompany(&forms.BillingCompany{
		CompanyKey:     config.Get().GetString(config.DefaultBillingCompanyKey),
		Name:           "PINGCAP",
		PaymentSubCode: "",
	}))
	datainit.CreateBillingConfig(tenantID, entity.BillingConfigType_DirectCustomer)

	for i, amt := range amounts {
		st := entity.InvoiceStatus_GENERATED
		if statuses != nil && i < len(statuses) {
			st = statuses[i]
		}
		inv := &entity.Invoice{
			TenantID:              tenantID,
			Status:                st,
			Type:                  entity.InvoiceType_COMMITMENT,
			InvoiceDate:           time.Date(2026, 8, 1+i, 0, 0, 0, 0, time.UTC),
			CostPrice:             decimal.NewFromFloat(amt),
			RawPrice:              decimal.NewFromFloat(amt),
			BindedBillingConfigID: billingConfigID,
		}
		assert.NoError(inv.Create(db.GetDB()))
	}
}

func (s *StripePayingWorkerSuite) runForMonth(month string) PayingResult {
	payingModel := &payment.PayingModel{}
	paying, err := payingModel.CreatePaying(month, "user:test", false)
	s.Require().NoError(err)

	// Manually claim to running (worker entry point does this; the
	// test drives runPaying directly).
	now := time.Now()
	s.Require().NoError(db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", paying.ID).
		Updates(map[string]interface{}{
			"status":       payment.PayingStatusRunning,
			"picked_up_at": now,
			"updated_at":   now,
		}).Error)

	invoiceModel := &payment.InvoiceModel{}
	start, end, _ := monthRange(month)
	result, err := runPaying(context.Background(), paying, start, end, invoiceModel, payingModel)
	s.Require().NoError(err)
	return result
}

// TestRunPayingPerInvoiceHappyPath: three above-threshold invoices,
// no carryover. All three go per-invoice via Pay.
func (s *StripePayingWorkerSuite) TestRunPayingPerInvoiceHappyPath() {
	assert := s.Assert()
	const tenantID uint = 1
	billingConfigID := uint(1)
	s.seedGroup(tenantID, billingConfigID, []float64{29.97, 100.00, 300.00}, nil)

	result := s.runForMonth("2026-08")

	assert.Len(result.Confirmed, 3, "all three invoices pushed")
	assert.Len(result.CarryoverUpdated, 0)
	assert.Len(result.Skipped, 0)
	// First crossing invoice triggers the merged push (single-month auto);
	// the remaining two go per-invoice.
	assert.Len(s.provider.PushInvoiceCalls, 1, "one merged push (first crossing)")
	assert.Len(s.provider.PayCalls, 2, "two per-invoice Pay calls")

	// All invoices must be SENT in the DB.
	var count int64
	db.GetDB().Model(&entity.Invoice{}).
		Where("tenant_id = ? AND status = ?", tenantID, entity.InvoiceStatus_SENT).
		Count(&count)
	assert.Equal(int64(3), count, "all invoices flipped to SENT")
}

// TestRunPayingCarryBelowThreshold: two below-threshold invoices,
// no crossing invoice. Both carried, none pushed.
func (s *StripePayingWorkerSuite) TestRunPayingCarryBelowThreshold() {
	assert := s.Assert()
	const tenantID uint = 2
	billingConfigID := uint(2)
	s.seedGroup(tenantID, billingConfigID, []float64{0.11, 0.71}, nil)

	result := s.runForMonth("2026-08")

	assert.Len(result.Confirmed, 0, "nothing pushed")
	assert.Len(result.CarryoverUpdated, 2, "both carried")
	assert.Empty(s.provider.PayCalls, "no Pay calls")
	assert.Empty(s.provider.PushInvoiceCalls, "no merged push")

	var count int64
	db.GetDB().Model(&entity.Invoice{}).
		Where("tenant_id = ? AND status = ?", tenantID, entity.InvoiceStatus_BELOW_THRESHOLD_CARRY).
		Count(&count)
	assert.Equal(int64(2), count, "both invoices below_threshold_carry")
}

// TestRunPayingCarryThenCrossThenPerInvoice: the canonical flow.
// $0.11 + $0.71 carried, $100 crosses (merged push with carried), $300
// per-invoice.
func (s *StripePayingWorkerSuite) TestRunPayingCarryThenCrossThenPerInvoice() {
	assert := s.Assert()
	const tenantID uint = 3
	billingConfigID := uint(3)

	// Seed a carried invoice from a previous month.
	carried := &entity.Invoice{
		TenantID:              tenantID,
		Status:                entity.InvoiceStatus_BELOW_THRESHOLD_CARRY,
		Type:                  entity.InvoiceType_COMMITMENT,
		InvoiceDate:           time.Date(2026, 7, 5, 0, 0, 0, 0, time.UTC),
		CostPrice:             decimal.NewFromFloat(0.11),
		RawPrice:              decimal.NewFromFloat(0.11),
		BindedBillingConfigID: billingConfigID,
	}
	assert.NoError(carried.Create(db.GetDB()))

	// Current month invoices.
	s.seedGroup(tenantID, billingConfigID, []float64{0.71, 100.00, 300.00}, nil)

	result := s.runForMonth("2026-08")

	// Expected: 0.11 (carried, accumulates), 0.71 (new, below → carry),
	// 100 (crossing → merged push with 0.11), 300 (per-invoice).
	assert.Len(result.CarryoverUpdated, 1, "0.71 carried")
	assert.Len(result.Confirmed, 2, "merged push + per-invoice")
	assert.Len(s.provider.PushInvoiceCalls, 1, "one merged push")
	assert.Len(s.provider.PayCalls, 1, "one per-invoice Pay ($300)")

	// Merged push batch = [carried 0.11, crossing 100].
	batch := s.provider.PushInvoiceCalls[0]
	assert.Len(batch, 2, "merged push bundles carried + crossing")
	assert.Equal(carried.ID, batch[0], "first batch item is the carried invoice")
	assert.Len(s.provider.PayCalls, 1, "one per-invoice Pay ($300)")
	// $300 invoice is the last seeded one; find its ID from the DB.
	var threeHundred entity.Invoice
	db.GetDB().Where("tenant_id = ? AND cost_price = ?", tenantID, decimal.NewFromFloat(300.00)).First(&threeHundred)
	assert.Equal(threeHundred.ID, s.provider.PayCalls[0], "Pay called with the $300 invoice")

	// DB statuses: carried + crossing → SENT, 0.71 → below_threshold_carry, 300 → SENT.
	var sent, carry int64
	db.GetDB().Model(&entity.Invoice{}).Where("tenant_id = ? AND status = ?", tenantID, entity.InvoiceStatus_SENT).Count(&sent)
	db.GetDB().Model(&entity.Invoice{}).Where("tenant_id = ? AND status = ?", tenantID, entity.InvoiceStatus_BELOW_THRESHOLD_CARRY).Count(&carry)
	assert.Equal(int64(3), sent, "carried + crossing + 300 all SENT")
	assert.Equal(int64(1), carry, "0.71 stays carried")
}

// TestRunPayingCarriedAloneCrosses: carried invoice alone crosses the
// threshold — merged push of just the carried batch.
func (s *StripePayingWorkerSuite) TestRunPayingCarriedAloneCrosses() {
	assert := s.Assert()
	const tenantID uint = 4
	billingConfigID := uint(4)

	// Two carried invoices totaling above $1.
	for i, amt := range []float64{0.60, 0.50} {
		carried := &entity.Invoice{
			TenantID:              tenantID,
			Status:                entity.InvoiceStatus_BELOW_THRESHOLD_CARRY,
			Type:                  entity.InvoiceType_COMMITMENT,
			InvoiceDate:           time.Date(2026, 7, 3+i, 0, 0, 0, 0, time.UTC),
			CostPrice:             decimal.NewFromFloat(amt),
			RawPrice:              decimal.NewFromFloat(amt),
			BindedBillingConfigID: billingConfigID,
		}
		assert.NoError(carried.Create(db.GetDB()))
	}

	s.seedGroup(tenantID, billingConfigID, []float64{29.97}, nil)

	result := s.runForMonth("2026-08")

	// Carried ($0.60+$0.50) already crosses the threshold; the first new
	// invoice ($29.97) joins the same merged push. One merged push of
	// three invoices, no per-invoice Pay.
	assert.Len(s.provider.PushInvoiceCalls, 1, "carried batch merged-pushed")
	assert.Len(s.provider.PushInvoiceCalls[0], 3, "carried + first new invoice bundled")
	assert.Len(s.provider.PayCalls, 0, "no per-invoice Pay")
	assert.Len(result.Confirmed, 1, "one merged push")
}

// TestRunPayingCarriedLaterThanNew reproduces the bug where a carried
// invoice sorted AFTER a new invoice (both carried earlier this month)
// crossed the threshold alone — pushing only the carried invoice and
// leaving the new one carried.
func (s *StripePayingWorkerSuite) TestRunPayingCarriedLaterThanNew() {
	assert := s.Assert()
	const tenantID uint = 5
	billingConfigID := uint(5)

	// Carried invoice with a LATER date (carried earlier this month).
	carried := &entity.Invoice{
		TenantID:              tenantID,
		Status:                entity.InvoiceStatus_BELOW_THRESHOLD_CARRY,
		Type:                  entity.InvoiceType_COMMITMENT,
		InvoiceDate:           time.Date(2026, 8, 20, 0, 0, 0, 0, time.UTC),
		CostPrice:             decimal.NewFromFloat(0.99),
		RawPrice:              decimal.NewFromFloat(0.99),
		BindedBillingConfigID: billingConfigID,
	}
	assert.NoError(carried.Create(db.GetDB()))

	// New invoice with an EARLIER date.
	newInv := &entity.Invoice{
		TenantID:              tenantID,
		Status:                entity.InvoiceStatus_GENERATED,
		Type:                  entity.InvoiceType_COMMITMENT,
		InvoiceDate:           time.Date(2026, 8, 10, 0, 0, 0, 0, time.UTC),
		CostPrice:             decimal.NewFromFloat(0.33),
		RawPrice:              decimal.NewFromFloat(0.33),
		BindedBillingConfigID: billingConfigID,
	}
	s.seedGroup(tenantID, billingConfigID, nil, nil)
	assert.NoError(newInv.Create(db.GetDB()))

	result := s.runForMonth("2026-08")

	// Expected: carried 0.99 + new 0.33 = 1.32 crosses → merged push of BOTH.
	assert.Len(s.provider.PushInvoiceCalls, 1, "one merged push")
	assert.Len(s.provider.PushInvoiceCalls[0], 2, "merged push bundles carried + new")
	assert.Len(result.Confirmed, 1, "one merged Confirmed")
	assert.Len(result.CarryoverUpdated, 0, "nothing newly carried")
	assert.Len(s.provider.PayCalls, 0, "no per-invoice Pay")

	// Both must be SENT in the DB.
	var sent int64
	db.GetDB().Model(&entity.Invoice{}).Where("tenant_id = ? AND status = ?", tenantID, entity.InvoiceStatus_SENT).Count(&sent)
	assert.Equal(int64(2), sent, "both carried and new flipped to SENT")
}

// mockCentralWithPermanentFreeFlag installs a CentralHandler that reports
// `permanentFreeTenantIDs` as hitting the billing_permenent_free_orgs flag
// (other tenants return false). Use this to drive the paying worker's
// defense-in-depth permanent-free filter without standing up a real central
// gRPC connection.
type mockPermanentFreeCentral struct {
	*MockCentralHandlerImpl
	permanentFreeTenantIDs map[uint64]bool
	flagCheckError         error
}

func (m *mockPermanentFreeCentral) IsHitFeatureFlag(orgID uint64, flag string) (bool, error) {
	if m.flagCheckError != nil {
		return false, m.flagCheckError
	}
	return m.permanentFreeTenantIDs[orgID], nil
}

// TestRunPayingSkipsPermanentFreeTenant: a tenant currently flagged as
// permanent-free must be skipped by runPaying even when an above-threshold
// GENERATED invoice exists for them. Mirrors the design doc FAQ: the
// status filter is the primary safety net, but a flag-config error or
// generation-time race could leave a stray GENERATED row, so the paying
// worker re-checks.
func (s *StripePayingWorkerSuite) TestRunPayingSkipsPermanentFreeTenant() {
	assert := s.Assert()
	const tenantID uint = 999
	billingConfigID := uint(999)

	central := &mockPermanentFreeCentral{
		MockCentralHandlerImpl: &MockCentralHandlerImpl{},
		permanentFreeTenantIDs: map[uint64]bool{uint64(tenantID): true},
	}
	central.SetAllTenants([]forms.Tenant{{TenantID: tenantID, Status: centralhandler.TenantStatusActive}})
	centralhandler.SetCentralHandler(central)

	s.seedGroup(tenantID, billingConfigID, []float64{100.00}, nil)

	result := s.runForMonth("2026-08")

	assert.Len(result.Confirmed, 0, "permanent-free tenant skipped — nothing pushed")
	assert.Len(result.CarryoverUpdated, 0)
	assert.Empty(s.provider.PushInvoiceCalls, "no merged push")
	assert.Empty(s.provider.PayCalls, "no per-invoice Pay")

	// Invoice must remain GENERATED in DB (we don't touch its status — that's
	// CreateInvoiceWorker's job).
	var generated int64
	db.GetDB().Model(&entity.Invoice{}).
		Where("tenant_id = ? AND status = ?", tenantID, entity.InvoiceStatus_GENERATED).
		Count(&generated)
	assert.Equal(int64(1), generated, "invoice untouched, still GENERATED")
}

// TestRunPayingOnlyAffectsFlaggedTenant: permanent-free flag for tenant A
// must not bleed into tenant B in the same run.
func (s *StripePayingWorkerSuite) TestRunPayingOnlyAffectsFlaggedTenant() {
	assert := s.Assert()

	central := &mockPermanentFreeCentral{
		MockCentralHandlerImpl: &MockCentralHandlerImpl{},
		permanentFreeTenantIDs: map[uint64]bool{uint64(777): true}, // 888 NOT flagged
	}
	central.SetAllTenants([]forms.Tenant{
		{TenantID: 777, Status: centralhandler.TenantStatusActive},
		{TenantID: 888, Status: centralhandler.TenantStatusActive},
	})
	centralhandler.SetCentralHandler(central)

	s.seedGroup(777, 777, []float64{100.00}, nil)
	// seedGroup creates a BillingCompany with the default key; the second
	// tenant needs its own to avoid the unique-key collision.
	billingInfoModel := &paymentsub.BillingInformationModel{}
	assert.NoError(billingInfoModel.CreateBillingCompany(&forms.BillingCompany{
		CompanyKey:     "tenant-888-co",
		Name:           "PINGCAP-888",
		PaymentSubCode: "",
	}))
	datainit.CreateBillingConfig(888, entity.BillingConfigType_DirectCustomer)
	paymentModel := &paymentsub.PaymentModel{}
	bcID888 := uint(888)
	_, err := paymentModel.CreatePaymentMethodAndActive(888, &forms.PaymentMethodCreateQuery{
		PaymentType:           entity.PaymentType_STRIPE,
		BindedBillingConfigID: &bcID888,
	})
	assert.NoError(err)
	inv := &entity.Invoice{
		TenantID:              888,
		Status:                entity.InvoiceStatus_GENERATED,
		Type:                  entity.InvoiceType_COMMITMENT,
		InvoiceDate:           time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC),
		CostPrice:             decimal.NewFromFloat(100.00),
		RawPrice:              decimal.NewFromFloat(100.00),
		BindedBillingConfigID: 888,
	}
	assert.NoError(inv.Create(db.GetDB()))

	result := s.runForMonth("2026-08")

	assert.Len(result.Confirmed, 1, "only non-flagged tenant 888 pushed")
	assert.Len(s.provider.PushInvoiceCalls, 1)
}

// TestTenantIsPermanentFreeFailsOpen: when the feature-flag RPC errors,
// tenantIsPermanentFree must return false so the paying worker continues
// rather than freezing the entire run. The status filter (SUSPENDED) is
// the primary safety net.
func (s *StripePayingWorkerSuite) TestTenantIsPermanentFreeFailsOpen() {
	central := &mockPermanentFreeCentral{
		MockCentralHandlerImpl: &MockCentralHandlerImpl{},
		flagCheckError:         fmt.Errorf("central gRPC unavailable"),
	}
	centralhandler.SetCentralHandler(central)

	assert := s.Assert()
	assert.False(tenantIsPermanentFree(context.Background(), 123),
		"flag-check error must fail open (return false)")
}