package tests

import (
	"errors"
	"testing"
	"time"

	"github.com/pingcap/dbaas/billing2/db"
	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/model/payment"
	"github.com/pingcap/dbaas/billing2/utils/tests"
	"github.com/shopspring/decimal"
	"github.com/stretchr/testify/suite"
)

// carryoverModel is shared across all carryover tests.
var carryoverModel = &payment.CarryoverModel{}

// CarryoverSuite covers the model layer of stripe_invoice_push_carryover:
// the optimistic-lock upsert, the source-invoice clear path, the summary
// rollup, and the 12-month expiry query.
type CarryoverSuite struct {
	tests.BillingTestSuite
}

func TestCarryoverSuite(t *testing.T) {
	suite.Run(t, new(CarryoverSuite))
}

// seedMonth is the canonical first_carry_month for fresh rows in this suite.
var seedMonth = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)

// TestUpsertWithAmountNewEntry verifies the create path: a row that did not
// exist is inserted with the merged amount and the JSON source list.
func (s *CarryoverSuite) TestUpsertWithAmountNewEntry() {
	assert := s.Assert()
	err := carryoverModel.UpsertWithAmount(1, 1, decimal.NewFromFloat(1.50), []uint{1001, 1002}, "2026-01")
	assert.NoError(err)

	row, err := carryoverModel.GetOrCreate(1, 1, seedMonth)
	assert.NoError(err)
	assert.Equal(decimal.NewFromFloat(1.50), row.Amount)
	assert.Equal(1, row.Version)

	items, err := payment.DecodeCarryoverSourceItemsForTest(row.SourceInvoiceIDs)
	assert.NoError(err)
	assert.Len(items, 2)
	assert.Equal(uint(1001), items[0].InvoiceID)
	assert.Equal(uint(1002), items[1].InvoiceID)
	assert.Equal("2026-01", items[0].Month)
}

// TestUpsertWithAmountUpdatesExisting verifies that a second upsert overwrites
// the existing row's amount (NOT additive) and bumps the version by 1. The
// second call's `amount` argument is the new total pool value, so the row's
// Amount column equals the second call's amount, not the sum.
func (s *CarryoverSuite) TestUpsertWithAmountUpdatesExisting() {
	assert := s.Assert()
	assert.NoError(carryoverModel.UpsertWithAmount(1, 1, decimal.NewFromFloat(0.50), []uint{1001}, "2026-01"))
	assert.NoError(carryoverModel.UpsertWithAmount(1, 1, decimal.NewFromFloat(0.30), []uint{1002}, "2026-02"))

	row, err := carryoverModel.GetOrCreate(1, 1, seedMonth)
	assert.NoError(err)
	// Overwrite semantics: row.Amount equals the second call's amount (0.30),
	// NOT the additive sum (0.80). This is the fix for the double-count bug.
	assert.Equal(decimal.NewFromFloat(0.30), row.Amount)
	assert.Equal(2, row.Version, "version must increment on each merge")

	items, err := payment.DecodeCarryoverSourceItemsForTest(row.SourceInvoiceIDs)
	assert.NoError(err)
	assert.Len(items, 2)
}

// TestUpsertWithAmountOverwritesExisting covers the double-count fix:
// a second upsert on an existing row must overwrite the amount, not add
// to it. After the fix, Amount = 2.10 (the second call's value), NOT 3.40
// (which would be row.Amount(1.30) + newAmount(2.10) under additive
// semantics). This is the regression test for Bug 2 in the carryover fix.
func (s *CarryoverSuite) TestUpsertWithAmountOverwritesExisting() {
	assert := s.Assert()
	// Seed: amount=1.30 with 2 invoice ids.
	assert.NoError(carryoverModel.UpsertWithAmount(6, 6, decimal.NewFromFloat(1.30), []uint{5001, 5002}, "2026-02"))

	// Caller passes the pre-summed "total" (carry.Amount + current month sum
	// = 1.30 + 0.80 = 2.10) — see stripe_invoice_paying_worker.runPaying.
	assert.NoError(carryoverModel.UpsertWithAmount(6, 6, decimal.NewFromFloat(2.10), []uint{5003}, "2026-03"))

	row, err := carryoverModel.GetOrCreate(6, 6, seedMonth)
	assert.NoError(err)
	// Overwrite semantics: the new amount is exactly the second call's value,
	// NOT the additive sum (1.30 + 2.10 = 3.40).
	assert.Equal(decimal.NewFromFloat(2.10), row.Amount, "amount must equal the second call's argument under overwrite semantics")
	assert.False(row.Amount.Equal(decimal.NewFromFloat(3.40)), "amount must NOT be the additive sum of both calls")
}

// TestUpsertWithAmountKeepsEarliestFirstCarryMonth verifies that
// first_carry_month is set to the earliest month we have ever seen.
func (s *CarryoverSuite) TestUpsertWithAmountKeepsEarliestFirstCarryMonth() {
	assert := s.Assert()
	// Seed with a later first_carry_month by inserting directly.
	row := &entity.StripeInvoicePushCarryover{
		TenantID:         5,
		BillingConfigID:  5,
		Amount:           decimal.Zero,
		SourceInvoiceIDs: "[]",
		FirstCarryMonth:  time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC),
		Version:          0,
	}
	assert.NoError(db.GetDB().Create(row).Error)

	assert.NoError(carryoverModel.UpsertWithAmount(5, 5, decimal.NewFromFloat(0.50), []uint{2001}, "2026-02"))
	reloaded, err := carryoverModel.GetOrCreate(5, 5, seedMonth)
	assert.NoError(err)
	// The earlier of (seed 2026-05, batch 2026-02) wins.
	assert.Equal(time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC), reloaded.FirstCarryMonth)
}

// TestUpsertWithAmountVersionConflict simulates two workers reading the same
// version; the loser must observe RowsAffected=0 and the helper returns an
// error after exhausting the retry budget. To force a conflict we hijack the
// retry budget by reading the row inside a hook on the *first* retry — but a
// simpler check is: bump the version between our read and our write, then
// verify the helper did not silently lose the increment.
//
// We assert correctness by checking that after the upsert completes the row
// reflects the latest state, and that no audit-data is lost. A "winning"
// upsert always converges; the version-conflict path is exercised in
// production by two workers racing, but in a single-threaded unit test the
// retry loop transparently absorbs the bump. So instead of asserting an error
// return, we verify the convergence: row.Amount reflects the second upsert.
func (s *CarryoverSuite) TestUpsertWithAmountVersionConflict() {
	assert := s.Assert()
	row, err := carryoverModel.GetOrCreate(7, 7, seedMonth)
	assert.NoError(err)
	originalVersion := row.Version

	// Simulate a concurrent writer by bumping the version out-of-band and
	// then making another upsert. The retry loop will read the new version
	// and re-apply the merge, so the end state must reflect the latest call.
	assert.NoError(db.GetDB().Model(&entity.StripeInvoicePushCarryover{}).
		Where("id = ?", row.ID).
		Update("version", originalVersion+5).Error)

	// This call MUST converge (no error) because the retry budget re-reads
	// after a stale write.
	assert.NoError(carryoverModel.UpsertWithAmount(7, 7, decimal.NewFromFloat(0.10), []uint{3001}, "2026-01"))

	reloaded, err := carryoverModel.GetOrCreate(7, 7, seedMonth)
	assert.NoError(err)
	assert.Equal(decimal.NewFromFloat(0.10).String(), reloaded.Amount.String())
}

// TestClearSourceInvoices verifies only the listed IDs are removed, the
// rest of the array is preserved, and the amount column is recomputed from
// the surviving items (so it stays in sync with source_invoice_ids).
func (s *CarryoverSuite) TestClearSourceInvoices() {
	assert := s.Assert()
	assert.NoError(carryoverModel.UpsertWithAmount(2, 2, decimal.NewFromFloat(1.20), []uint{1001, 1002, 1003}, "2026-03"))

	assert.NoError(carryoverModel.ClearSourceInvoices(2, 2, []uint{1002}))
	row, err := carryoverModel.GetOrCreate(2, 2, seedMonth)
	assert.NoError(err)

	items, err := payment.DecodeCarryoverSourceItemsForTest(row.SourceInvoiceIDs)
	assert.NoError(err)
	assert.Len(items, 2)
	assert.Equal(uint(1001), items[0].InvoiceID)
	assert.Equal(uint(1003), items[1].InvoiceID)

	// buildSourceItems splits 1.20 evenly across 3 items (each=0.40, sum
	// = 0.40 + 0.40 + 0.40 = 1.20). Removing one item leaves 0.40 + 0.40
	// = 0.80. The amount column must reflect that, not the stale 1.20.
	assert.Equal(decimal.NewFromFloat(0.80), row.Amount)
}

// TestClearSourceInvoicesResetsAmount covers the empty-pool scenario: when
// every source invoice ID is cleared, the row's amount must drop to zero
// so the dashboard's carryover_summary.total_amount does not report a
// stale non-zero total for an empty pool.
func (s *CarryoverSuite) TestClearSourceInvoicesResetsAmount() {
	assert := s.Assert()
	assert.NoError(carryoverModel.UpsertWithAmount(8, 8, decimal.NewFromFloat(1.30), []uint{4001, 4002}, "2026-02"))

	assert.NoError(carryoverModel.ClearSourceInvoices(8, 8, []uint{4001, 4002}))
	row, err := carryoverModel.GetOrCreate(8, 8, seedMonth)
	assert.NoError(err)

	items, err := payment.DecodeCarryoverSourceItemsForTest(row.SourceInvoiceIDs)
	assert.NoError(err)
	assert.Empty(items, "source_invoice_ids must be empty after clearing every ID")
	assert.True(row.Amount.IsZero(), "amount must be zero after clearing every ID, got %s", row.Amount)
}

// TestClearSourceInvoicesNoopOnEmpty verifies that an empty consumed list is
// a no-op (no version bump).
func (s *CarryoverSuite) TestClearSourceInvoicesNoopOnEmpty() {
	assert := s.Assert()
	assert.NoError(carryoverModel.UpsertWithAmount(2, 3, decimal.NewFromFloat(0.40), []uint{1001}, "2026-03"))

	before, err := carryoverModel.GetOrCreate(2, 3, seedMonth)
	assert.NoError(err)
	assert.NoError(carryoverModel.ClearSourceInvoices(2, 3, nil))
	after, err := carryoverModel.GetOrCreate(2, 3, seedMonth)
	assert.NoError(err)
	assert.Equal(before.Version, after.Version, "empty clear must not bump the version")
}

// TestCountAggregatesCorrectly verifies total_count and total_amount.
func (s *CarryoverSuite) TestCountAggregatesCorrectly() {
	assert := s.Assert()
	assert.NoError(carryoverModel.UpsertWithAmount(3, 1, decimal.NewFromFloat(0.30), []uint{1}, "2026-01"))
	assert.NoError(carryoverModel.UpsertWithAmount(3, 2, decimal.NewFromFloat(0.50), []uint{2}, "2026-01"))
	assert.NoError(carryoverModel.UpsertWithAmount(3, 3, decimal.NewFromFloat(0.20), []uint{3}, "2026-01"))

	count, total, err := carryoverModel.Count()
	assert.NoError(err)
	assert.Equal(int64(3), count)
	// Pass decimal.Decimal directly so the BillingAssert.Equal helper
	// compares numerically (0.30 + 0.50 + 0.20 = 1.0). String comparison
	// would fail because decimal.String() drops trailing zeros.
	assert.Equal(decimal.NewFromFloat(1.0), total)
}

// TestSummary verifies the dashboard rollup: count, total amount, and the
// earliest first_carry_month.
func (s *CarryoverSuite) TestSummary() {
	assert := s.Assert()
	// Three rows: first_carry_month 2026-01, 2026-02, 2026-03.
	rowA := &entity.StripeInvoicePushCarryover{
		TenantID: 10, BillingConfigID: 1, Amount: decimal.NewFromFloat(0.50),
		SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC),
	}
	rowB := &entity.StripeInvoicePushCarryover{
		TenantID: 10, BillingConfigID: 2, Amount: decimal.NewFromFloat(0.30),
		SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC),
	}
	rowC := &entity.StripeInvoicePushCarryover{
		TenantID: 10, BillingConfigID: 3, Amount: decimal.NewFromFloat(0.20),
		SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC),
	}
	assert.NoError(db.GetDB().Create(rowA).Error)
	assert.NoError(db.GetDB().Create(rowB).Error)
	assert.NoError(db.GetDB().Create(rowC).Error)

	summary, err := carryoverModel.Summary()
	assert.NoError(err)
	assert.Equal(int64(3), summary.TotalCount)
	assert.Equal(decimal.NewFromFloat(1.0), summary.TotalAmount)
	assert.NotNil(summary.OldestFirstCarryMonth)
	assert.Equal(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), *summary.OldestFirstCarryMonth)
}

// TestSummaryEmptyPool ensures OldestFirstCarryMonth is nil for an empty
// pool so the API can distinguish "no carryovers" from "January 0001".
func (s *CarryoverSuite) TestSummaryEmptyPool() {
	assert := s.Assert()
	summary, err := carryoverModel.Summary()
	assert.NoError(err)
	assert.Equal(int64(0), summary.TotalCount)
	assert.True(summary.TotalAmount.IsZero())
	assert.Nil(summary.OldestFirstCarryMonth)
}

// TestFindExpired verifies the 12-month cap query: rows older than the
// threshold are returned, newer ones are excluded.
func (s *CarryoverSuite) TestFindExpired() {
	assert := s.Assert()
	threshold := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC)

	old1 := &entity.StripeInvoicePushCarryover{
		TenantID: 20, BillingConfigID: 1, Amount: decimal.NewFromFloat(0.30),
		SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC),
	}
	old2 := &entity.StripeInvoicePushCarryover{
		TenantID: 20, BillingConfigID: 2, Amount: decimal.NewFromFloat(0.50),
		SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2025, 5, 1, 0, 0, 0, 0, time.UTC),
	}
	fresh := &entity.StripeInvoicePushCarryover{
		TenantID: 20, BillingConfigID: 3, Amount: decimal.NewFromFloat(0.10),
		SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC),
	}
	assert.NoError(db.GetDB().Create(old1).Error)
	assert.NoError(db.GetDB().Create(old2).Error)
	assert.NoError(db.GetDB().Create(fresh).Error)

	expired, err := carryoverModel.FindExpired(threshold)
	assert.NoError(err)
	assert.Len(expired, 2)
	for _, row := range expired {
		assert.True(row.FirstCarryMonth.Before(threshold),
			"row first_carry_month=%s must be < threshold %s", row.FirstCarryMonth, threshold)
	}
}

// TestListOldestFirstCarryMonth verifies the oldest-first ordering used by
// the dashboard's old-month list.
func (s *CarryoverSuite) TestListOldestFirstCarryMonth() {
	assert := s.Assert()
	rows := []*entity.StripeInvoicePushCarryover{
		{TenantID: 30, BillingConfigID: 1, Amount: decimal.NewFromFloat(0.10), SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC)},
		{TenantID: 30, BillingConfigID: 2, Amount: decimal.NewFromFloat(0.10), SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2025, 11, 1, 0, 0, 0, 0, time.UTC)},
		{TenantID: 30, BillingConfigID: 3, Amount: decimal.NewFromFloat(0.10), SourceInvoiceIDs: "[]", FirstCarryMonth: time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC)},
	}
	for _, r := range rows {
		assert.NoError(db.GetDB().Create(r).Error)
	}

	got, err := carryoverModel.ListOldestFirstCarryMonth(2)
	assert.NoError(err)
	assert.Len(got, 2)
	assert.Equal(time.Date(2025, 11, 1, 0, 0, 0, 0, time.UTC), got[0].FirstCarryMonth)
	assert.Equal(time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC), got[1].FirstCarryMonth)
}

// TestClearSourceInvoicesWithDBInsideTransaction verifies the WithDB
// variant participates in an outer transaction. We seed a carryover row,
// run ClearSourceInvoicesWithDB inside a tx that we then commit, and
// assert the source_invoice_ids change took effect. Combined with the
// outer transaction behaviour of db.GetDB().Transaction itself, this
// proves that the WithDB variant does not open its own transaction.
//
// We cannot directly assert rollback semantics in a unit test (would
// require forcing a mid-call conflict, which the helper's retry loop
// self-heals). Instead, this test pins the happy-path contract: a
// successful WithDB call inside an outer tx commits atomically.
func (s *CarryoverSuite) TestClearSourceInvoicesWithDBInsideTransaction() {
	assert := s.Assert()
	assert.NoError(carryoverModel.UpsertWithAmount(70, 71, decimal.NewFromFloat(0.80), []uint{8001, 8002}, "2026-04"))

	tx := db.GetDB().Begin()
	assert.NoError(tx.Error)
	assert.NoError(carryoverModel.ClearSourceInvoicesWithDB(tx, 70, 71, []uint{8001}))
	assert.NoError(tx.Commit().Error)

	reloaded, err := carryoverModel.GetOrCreate(70, 71, seedMonth)
	assert.NoError(err)
	items, err := payment.DecodeCarryoverSourceItemsForTest(reloaded.SourceInvoiceIDs)
	assert.NoError(err)
	assert.Len(items, 1, "8001 must be cleared inside the committed tx")
	assert.Equal(uint(8002), items[0].InvoiceID)
}

// TestUpsertWithAmountWithDBInsideTransaction pins the happy-path contract of
// the transactional variant: it must write through the caller's tx (no
// self-opened transaction) and commit atomically, mirroring
// TestClearSourceInvoicesWithDBInsideTransaction for the upsert path.
func (s *CarryoverSuite) TestUpsertWithAmountWithDBInsideTransaction() {
	assert := s.Assert()
	tx := db.GetDB().Begin()
	assert.NoError(tx.Error)
	err := carryoverModel.UpsertWithAmountWithDB(tx, 80, 81, decimal.NewFromFloat(1.20), []uint{9001, 9002}, "2026-04")
	assert.NoError(err)
	assert.NoError(tx.Commit().Error)

	reloaded, err := carryoverModel.GetOrCreate(80, 81, seedMonth)
	assert.NoError(err)
	assert.Equal(decimal.NewFromFloat(1.20), reloaded.Amount)
	items, err := payment.DecodeCarryoverSourceItemsForTest(reloaded.SourceInvoiceIDs)
	assert.NoError(err)
	assert.Len(items, 2, "upsert inside the committed tx must persist")
	assert.Equal(uint(9001), items[0].InvoiceID)
	assert.Equal(uint(9002), items[1].InvoiceID)
}

// TestUpsertWithAmountWithDBVersionConflictRollsBack is the regression test
// for the silent-drop bug: UpsertWithAmountWithDB used to return nil when the
// optimistic-lock UPDATE matched 0 rows (RowsAffected==0), so the worker's tx
// committed the BELOW_THRESHOLD_CARRY status flip while the pool write never
// landed — invoices were dropped from the billing pipeline with no pool entry
// and no error. The fixed helper must return a typed CarryoverVersionConflictError
// (which the worker's errors.As releases back to pending) and write nothing.
//
// The helper reads the carryover row inside the caller's tx (it no longer
// falls back to a separate-connection GetOrCreate), so a version bump inside
// the same tx would be visible to the helper's own read and no conflict would
// surface. To force the race deterministically we freeze the tx's consistent
// snapshot at version 1 with a dummy read, then commit a version bump on a
// separate connection. The helper's SELECT then serves the stale snapshot
// (version 1) while its UPDATE current-reads the latest committed version
// (99) — 0 rows matched and the typed conflict surfaces.
func (s *CarryoverSuite) TestUpsertWithAmountWithDBVersionConflictRollsBack() {
	assert := s.Assert()
	assert.NoError(carryoverModel.UpsertWithAmount(82, 83, decimal.NewFromFloat(0.50), []uint{9003}, "2026-01"))

	tx := db.GetDB().Begin()
	assert.NoError(tx.Error)
	// Freeze the tx's snapshot at the committed version (1) so the helper's
	// in-tx SELECT cannot observe the bump we commit below.
	var probe entity.StripeInvoicePushCarryover
	assert.NoError(tx.Where("tenant_id = ? AND billing_config_id = ?", 82, 83).First(&probe).Error)
	assert.Equal(1, probe.Version)

	// Simulate the concurrent writer: commit the version bump on a separate
	// connection while this tx still holds its stale snapshot.
	assert.NoError(db.GetDB().Model(&entity.StripeInvoicePushCarryover{}).
		Where("tenant_id = ? AND billing_config_id = ?", 82, 83).
		Update("version", 99).Error)

	err := carryoverModel.UpsertWithAmountWithDB(tx, 82, 83, decimal.NewFromFloat(1.00), []uint{9004}, "2026-02")
	var conflict *payment.CarryoverVersionConflictError
	assert.True(errors.As(err, &conflict), "conflict must surface as CarryoverVersionConflictError, got %v", err)
	assert.Equal("upsert", conflict.Op)
	assert.Equal(1, conflict.Attempts)

	// Roll back the tx (as the worker does on conflict) and verify the pool
	// row carries no trace of the conflicted upsert — the status flip that
	// would have been wrapped in the same tx is likewise rolled back. The
	// version is 99 (the simulated concurrent writer's committed bump), not
	// the loser's stale read: the loser's rollback must not clobber the
	// winner's committed write.
	assert.NoError(tx.Rollback().Error)
	reloaded, err := carryoverModel.GetOrCreate(82, 83, seedMonth)
	assert.NoError(err)
	assert.Equal(99, reloaded.Version, "the loser's rollback must leave the concurrent writer's version bump intact")
	assert.Equal(decimal.NewFromFloat(0.50), reloaded.Amount)
	items, err := payment.DecodeCarryoverSourceItemsForTest(reloaded.SourceInvoiceIDs)
	assert.NoError(err)
	assert.Len(items, 1)
	assert.Equal(uint(9003), items[0].InvoiceID, "the conflicted upsert must not leak items into the pool")
}

// TestCarryoverVersionConflictErrorType pins the typed error surface so
// callers can rely on errors.As. The retry-budget exhaustion path is
// covered by the existing TestUpsertWithAmountVersionConflict (which
// proves the retry loop converges under single-shot contention); the
// typed error is constructed by the helper on exhaustion, and we verify
// its construction here without needing to race the retry loop.
func (s *CarryoverSuite) TestCarryoverVersionConflictErrorType() {
	assert := s.Assert()
	base := errors.New("synthetic conflict")
	conflict := &payment.CarryoverVersionConflictError{
		Op:       "clear",
		Attempts: 5,
		Cause:    base,
	}
	assert.Equal("clear", conflict.Op)
	assert.Equal(5, conflict.Attempts)
	assert.True(errors.Is(conflict, base), "errors.Is must unwrap to the underlying cause")

	// The error message must encode both Op and Attempts so logs are
	// self-describing when the worker emits a transient_retry log line.
	assert.Contains(conflict.Error(), "clear")
	assert.Contains(conflict.Error(), "5")
}