package tests

import (
	"errors"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/pingcap/dbaas/billing2/billing_errors"
	"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/pingcap/dbaas/billing2/utils/timeutils"
	"github.com/stretchr/testify/suite"
	"gorm.io/gorm"
)

// payingModel is shared across all paying tests in this file.
var payingModel = &payment.PayingModel{}

// PayingSuite covers the model layer of stripe_invoice_paying: creation, the
// same-month uniqueness guarantee, the worker-side claim/complete/fail
// transitions, and the dashboard's recent-list query.
//
// Each test starts from an empty table (BillingTestSuite cleans between tests)
// because payingModel.CreatePaying relies on the uk_active_paying constraint.
type PayingSuite struct {
	tests.BillingTestSuite
}

func TestPayingSuite(t *testing.T) {
	suite.Run(t, new(PayingSuite))
}

// TestCreatePayingHappyPath verifies the basic insert path: pending status,
// correct month + createdBy. active_flag is a STORED GENERATED column
// derived from status, so it is verified by a fresh SELECT against the DB
// (GORM `->` skips write/read on Insert so the returned paying does not
// carry the generated value).
func (s *PayingSuite) TestCreatePayingHappyPath() {
	assert := s.Assert()
	paying, err := payingModel.CreatePaying("2026-06", "user:42", false)
	assert.NoError(err)
	assert.NotNil(paying)
	assert.NotZero(paying.ID)
	assert.Equal("2026-06", paying.Month)
	assert.Equal(payment.PayingStatusPending, paying.Status)
	assert.Equal("user:42", paying.CreatedBy)

	// Round-trip via GetPaying to verify the DB-computed values, including
	// the STORED GENERATED active_flag.
	reloaded, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(paying.ID, reloaded.ID)
	assert.Equal(paying.Status, reloaded.Status)
	assert.NotNil(reloaded.ActiveFlag, "pending row must have active_flag=1 from GENERATED column")
	assert.Equal(uint(1), *reloaded.ActiveFlag)
}

// TestCreatePayingRejectsSameMonthActive asserts the 409 path: a second
// pending row for the same month must be rejected with
// billing_errors.PayingAlreadyActiveError.
func (s *PayingSuite) TestCreatePayingRejectsSameMonthActive() {
	assert := s.Assert()
	first, err := payingModel.CreatePaying("2026-07", "user:1", false)
	assert.NoError(err)
	assert.NotNil(first)

	_, err = payingModel.CreatePaying("2026-07", "user:2", false)
	assert.Error(err)
	assert.True(isPayingAlreadyActive(err), "expected PayingAlreadyActiveError, got %v", err)
}

// TestCreatePayingAllowsAfterCompleted confirms that completed (or failed)
// rows do not block a fresh create. The DB unique key only enforces
// uniqueness for rows whose active_flag is non-NULL.
func (s *PayingSuite) TestCreatePayingAllowsAfterCompleted() {
	assert := s.Assert()
	first, err := payingModel.CreatePaying("2026-08", "user:1", false)
	assert.NoError(err)

	// Walk first through pending -> running -> completed.
	now := timeutils.GetNow()
	flag := uint(1)
	first.Status = payment.PayingStatusRunning
	first.PickedUpAt = &now
	first.ActiveFlag = &flag
	assert.NoError(db.GetDB().Save(first).Error)
	assert.NoError(payingModel.CompletePaying(first, `{"confirmed":[]}`))

	// A new pending row for the same month must succeed.
	second, err := payingModel.CreatePaying("2026-08", "user:2", false)
	assert.NoError(err)
	assert.NotNil(second)
	assert.NotEqual(first.ID, second.ID)
	assert.Equal(payment.PayingStatusPending, second.Status)
}

// TestCreatePayingAllowsAfterFailed is the symmetric check for failed rows.
func (s *PayingSuite) TestCreatePayingAllowsAfterFailed() {
	assert := s.Assert()
	first, err := payingModel.CreatePaying("2026-09", "user:1", false)
	assert.NoError(err)
	manuallyClaimRunning(s.T(), first.ID)
	assert.NoError(payingModel.FailPaying(first, "synthetic failure"))
	second, err := payingModel.CreatePaying("2026-09", "user:2", false)
	assert.NoError(err)
	assert.NotNil(second)
}

// TestCreatePayingForceReplacesActive verifies the Ops escape hatch:
// force=true wipes the existing pending/running row and inserts a fresh one
// in the same transaction, so the new row's active_flag never collides with
// the unique key.
//
// The old row must be physically removed from the table, not just soft-
// deleted: gorm's default Delete() sets deleted_at but leaves the row, and
// uk_active_paying would still fire on the next insert. We assert this with
// an Unscoped().First() — a regular First() would mask the regression by
// filtering the soft-deleted row out.
func (s *PayingSuite) TestCreatePayingForceReplacesActive() {
	assert := s.Assert()
	old, err := payingModel.CreatePaying("2026-10", "user:1", false)
	assert.NoError(err)

	replacement, err := payingModel.CreatePaying("2026-10", "user:ops", true)
	assert.NoError(err)
	assert.NotNil(replacement)
	assert.NotEqual(old.ID, replacement.ID)
	assert.Equal("user:ops", replacement.CreatedBy)
	assert.Equal(payment.PayingStatusPending, replacement.Status)

	// The old row must be physically gone — Unscoped() bypasses soft-delete
	// filtering so a passing assertion here proves the row was hard-deleted,
	// not merely marked with deleted_at.
	var remaining entity.StripeInvoicePaying
	err = db.GetDB().Unscoped().Where("id = ?", old.ID).First(&remaining).Error
	assert.True(isRecordNotFound(err), "old row should be hard-deleted (Unscoped), got err=%v", err)

	// And only one row should exist for this month now.
	reloaded, err := payingModel.ListActivePayingByMonth("2026-10")
	assert.NoError(err)
	assert.Equal(replacement.ID, reloaded.ID)
}

// TestCreatePayingForceKeepsCompleted verifies that force=true only deletes
// pending/running rows. Completed (and failed) rows are historical records
// and must survive.
func (s *PayingSuite) TestCreatePayingForceKeepsCompleted() {
	assert := s.Assert()
	completed, err := payingModel.CreatePaying("2026-11", "user:1", false)
	assert.NoError(err)
	now := timeutils.GetNow()
	flag := uint(1)
	completed.Status = payment.PayingStatusRunning
	completed.PickedUpAt = &now
	completed.ActiveFlag = &flag
	assert.NoError(db.GetDB().Save(completed).Error)
	assert.NoError(payingModel.CompletePaying(completed, `{"ok":true}`))

	// force=true on the same month: should succeed because completed rows
	// do not collide on uk_active_paying.
	replacement, err := payingModel.CreatePaying("2026-11", "user:ops", true)
	assert.NoError(err)
	assert.NotNil(replacement)

	// Completed row is still there with its original id.
	var archived entity.StripeInvoicePaying
	assert.NoError(db.GetDB().Where("id = ?", completed.ID).First(&archived).Error)
	assert.Equal(payment.PayingStatusCompleted, archived.Status)
	assert.Nil(archived.ActiveFlag)
}

// TestClaimNextPendingPaying verifies the atomic claim: oldest pending wins,
// status flips to running, picked_up_at is populated.
//
// Claim now uses a UUID marker matched via result LIKE '%<uuid>%' —
// no clock comparison. This test verifies the claim succeeds with
// a nanosecond-precision timestamp (the marker approach is immune to
// DATETIME(3) truncation).
func (s *PayingSuite) TestClaimNextPendingPaying() {
	assert := s.Assert()
	fixedNow := time.Date(2026, 6, 1, 12, 0, 0, 123456789, time.UTC)
	timeutils.SetMockNow(fixedNow)
	defer timeutils.SetMockNow(time.Time{})

	// Insert two pending rows out of created_at order to verify ORDER BY.
	first := &entity.StripeInvoicePaying{
		Month:      "2026-06",
		Status:     payment.PayingStatusPending,
		ActiveFlag: uintPtr(1),
		CreatedBy:  "user:1",
	}
	assert.NoError(db.GetDB().Create(first).Error)
	time.Sleep(10 * time.Millisecond)
	second := &entity.StripeInvoicePaying{
		Month:      "2026-07",
		Status:     payment.PayingStatusPending,
		ActiveFlag: uintPtr(1),
		CreatedBy:  "user:2",
	}
	assert.NoError(db.GetDB().Create(second).Error)

	claimed, err := payingModel.ClaimNextPendingPaying()
	assert.NoError(err)
	assert.NotNil(claimed)
	assert.Equal(first.ID, claimed.ID, "oldest pending must be claimed first")
	assert.Equal(payment.PayingStatusRunning, claimed.Status)
	assert.NotNil(claimed.PickedUpAt)
}

// TestClaimNextPendingPayingEmpty confirms nil/nil when there is no work.
func (s *PayingSuite) TestClaimNextPendingPayingEmpty() {
	assert := s.Assert()
	claimed, err := payingModel.ClaimNextPendingPaying()
	assert.NoError(err)
	assert.Nil(claimed)
}

// Documents the original bug (pre-UUID claim matched on picked_up_at < now,
// which DATETIME(3) truncation could break). With the UUID-marker claim
// this regression is structurally impossible, but the test remains as
// a sentinel.
func (s *PayingSuite) TestClaimNextPendingPayingPrecisionRegression() {
	assert := s.Assert()
	// 789ns — well below the ms boundary so the stored value is the same
	// instant as the Go value modulo a ~789ns truncation.
	fixedNow := time.Date(2026, 9, 15, 10, 30, 45, 789, time.UTC)
	timeutils.SetMockNow(fixedNow)
	defer timeutils.SetMockNow(time.Time{})

	pending, err := payingModel.CreatePaying("2026-09", "user:precision", false)
	assert.NoError(err)
	assert.NotNil(pending)

	claimed, err := payingModel.ClaimNextPendingPaying()
	assert.NoError(err)
	assert.NotNil(claimed, "claim must return the row despite the ns/ms truncation in picked_up_at")
	assert.Equal(pending.ID, claimed.ID)
	assert.Equal(payment.PayingStatusRunning, claimed.Status)
}

// TestClaimNextPendingPayingWritesClaimUUID verifies that the claim
// writes a deterministic UUID marker into the result column so the
// subsequent SELECT can match on it — no clock comparison, no
// DATETIME(3) truncation race.
//
// After CompletePaying the real result overwrites the marker; after
// FailPaying the marker remains visible for ops inspection.
func (s *PayingSuite) TestClaimNextPendingPayingWritesClaimUUID() {
	assert := s.Assert()
	pending, err := payingModel.CreatePaying("2027-01", "user:claim-uuid", false)
	assert.NoError(err)

	claimed, err := payingModel.ClaimNextPendingPaying()
	assert.NoError(err)
	assert.NotNil(claimed)
	assert.Equal(pending.ID, claimed.ID)
	assert.Equal(payment.PayingStatusRunning, claimed.Status)

	// Verify the claim marker is written in the result column.
	// MySQL JSON type normalizes {"key":"val"} to {"key": "val"} (space
	// after colon), so we check for the key substring.
	reloaded, err := payingModel.GetPaying(claimed.ID)
	assert.NoError(err)
	assert.NotNil(reloaded.Result, "result must contain the claim marker after claim")
	assert.True(strings.Contains(*reloaded.Result, `"_claim_id"`),
		"result must contain the claim marker, got %q", *reloaded.Result)

	// CompletePaying must overwrite the marker with the real result.
	// MySQL JSON column normalizes whitespace and key order, so we don't
	// compare exact strings; we just assert the result is non-nil and
	// the claim marker is gone.
	resultJSON := `{"confirmed":[{"tenant_id":1,"amount":"5.00","push_type":"single_month"}]}`
	assert.NoError(payingModel.CompletePaying(claimed, resultJSON))
	final, err := payingModel.GetPaying(claimed.ID)
	assert.NoError(err)
	assert.NotNil(final.Result)
	assert.NotContains(*final.Result, `"_claim_id"`,
		"result must not contain the claim marker after CompletePaying, got %q", *final.Result)
}

// TestClaimNextPendingPayingConcurrent verifies that when multiple
// workers race to claim the only pending row, exactly one wins. The
// UPDATE with LIMIT 1 guarantees that only one worker flips the row;
// every other worker sees RowsAffected == 0 and gets (nil, nil).
//
// The winning claim must carry the UUID marker; every losing claim
// must be nil.
func (s *PayingSuite) TestClaimNextPendingPayingConcurrent() {
	assert := s.Assert()
	pending, err := payingModel.CreatePaying("2027-02", "user:concurrent", false)
	assert.NoError(err)
	assert.NotNil(pending)

	const workers = 5
	var wg sync.WaitGroup
	var mu sync.Mutex
	var winnerIDs []uint
	var nilCount int

	for range workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			claimed, claimErr := payingModel.ClaimNextPendingPaying()
			assert.NoError(claimErr)
			mu.Lock()
			defer mu.Unlock()
			if claimed == nil {
				nilCount++
				return
			}
			winnerIDs = append(winnerIDs, claimed.ID)
		}()
	}
	wg.Wait()

	assert.Equal(1, len(winnerIDs),
		"exactly one worker must win the claim, got %d winners with IDs %v",
		len(winnerIDs), winnerIDs)
	assert.Equal(pending.ID, winnerIDs[0],
		"the winning claim must be the only pending row")
	assert.Equal(workers-1, nilCount,
		"all other workers must lose the race and get nil")

	// The winning claim's result column must carry a claim marker.
	winner, err := payingModel.GetPaying(winnerIDs[0])
	assert.NoError(err)
	assert.Equal(payment.PayingStatusRunning, winner.Status)
	assert.NotNil(winner.Result)
	assert.True(strings.Contains(*winner.Result, `"_claim_id"`),
		"winning claim must have a claim marker in result, got %q", *winner.Result)
}

// TestCompletePayingAndFailPaying covers the two terminal transitions.
func (s *PayingSuite) TestCompletePayingAndFailPaying() {
	assert := s.Assert()
	pending, err := payingModel.CreatePaying("2026-12", "user:1", false)
	assert.NoError(err)
	manuallyClaimRunning(s.T(), pending.ID)

	resultJSON := `{"confirmed":[{"tenant_id":1,"amount":"3.00","push_type":"single_month"}]}`
	assert.NoError(payingModel.CompletePaying(pending, resultJSON))
	assert.Equal(payment.PayingStatusCompleted, pending.Status)
	assert.NotNil(pending.CompletedAt)
	assert.NotNil(pending.Result)
	assert.Equal(resultJSON, *pending.Result)
	assert.Nil(pending.ActiveFlag, "active_flag must be cleared on completion")

	// Fail path on a separate row.
	failing, err := payingModel.CreatePaying("2027-01", "user:1", false)
	assert.NoError(err)
	manuallyClaimRunning(s.T(), failing.ID)
	assert.NoError(payingModel.FailPaying(failing, "stripe 502"))
	assert.Equal(payment.PayingStatusFailed, failing.Status)
	assert.Equal("stripe 502", failing.ErrorMessage)
	assert.Nil(failing.ActiveFlag)
}

// TestListRecentPayings verifies the dashboard ordering (newest first) and
// the limit boundary. We use three distinct months because uk_active_paying
// blocks same-month duplicates.
func (s *PayingSuite) TestListRecentPayings() {
	assert := s.Assert()
	months := []string{"2026-06", "2026-07", "2026-08"}
	for _, m := range months {
		_, err := payingModel.CreatePaying(m, "user:1", false)
		assert.NoError(err)
		time.Sleep(5 * time.Millisecond)
	}
	got, err := payingModel.ListRecentPayings(2)
	assert.NoError(err)
	assert.Len(got, 2)
	for i := 1; i < len(got); i++ {
		assert.True(!got[i-1].CreatedAt.Before(got[i].CreatedAt), "rows must be DESC by created_at")
	}
	// Most recent (2026-08) should be first.
	assert.Equal("2026-08", got[0].Month)
}

// TestUpdatePayingResult covers the worker-side partial-state write. MySQL
// normalises JSON values (it adds whitespace), so the assertion compares the
// parsed JSON value rather than the raw string.
func (s *PayingSuite) TestUpdatePayingResult() {
	assert := s.Assert()
	paying, err := payingModel.CreatePaying("2026-06", "user:1", false)
	assert.NoError(err)
	assert.Nil(paying.Result)

	assert.NoError(payingModel.UpdatePayingResult(paying.ID, `{"confirmed":[]}`))
	reloaded, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.NotNil(reloaded.Result)
	assert.NotEmpty(*reloaded.Result)
	// Status is not changed by an UpdatePayingResult call.
	assert.Equal(payment.PayingStatusPending, reloaded.Status)
}

// TestUpdatePayingResultMissing returns ObjectNotExistsError for an id that
// does not exist.
func (s *PayingSuite) TestUpdatePayingResultMissing() {
	assert := s.Assert()
	err := payingModel.UpdatePayingResult(99999999, "{}")
	assert.Error(err)
	assert.True(isObjectNotExists(err), "expected ObjectNotExistsError, got %v", err)
}

// TestActiveFlagGeneratedFromStatus is the contract test for the STORED
// GENERATED column stripe_invoice_paying.active_flag. The application does
// not write to active_flag — its value is computed by MySQL from status.
// This is what lets uk_active_paying(month, active_flag) reliably enforce
// "at most one active paying per month" without app-managed drift.
func (s *PayingSuite) TestActiveFlagGeneratedFromStatus() {
	assert := s.Assert()

	// pending -> active_flag=1
	pending, err := payingModel.CreatePaying("2027-02", "user:1", false)
	assert.NoError(err)
	reloaded, err := payingModel.GetPaying(pending.ID)
	assert.NoError(err)
	assert.NotNil(reloaded.ActiveFlag, "pending must yield active_flag=1")
	assert.Equal(uint(1), *reloaded.ActiveFlag)

	// running (worker claim) -> active_flag=1 (still 1, status still
	// in the "active" set)
	now := timeutils.GetNow()
	assert.NoError(db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", pending.ID).
		Updates(map[string]interface{}{
			"status":       payment.PayingStatusRunning,
			"picked_up_at": now,
		}).Error)
	reloaded, err = payingModel.GetPaying(pending.ID)
	assert.NoError(err)
	assert.NotNil(reloaded.ActiveFlag, "running must also yield active_flag=1")
	assert.Equal(uint(1), *reloaded.ActiveFlag)

	// completed -> active_flag=NULL
	assert.NoError(payingModel.CompletePaying(pending, `{"ok":true}`))
	reloaded, err = payingModel.GetPaying(pending.ID)
	assert.NoError(err)
	assert.Nil(reloaded.ActiveFlag, "completed must yield active_flag=NULL")

	// failed -> active_flag=NULL (separate row, same-month coexistence)
	failing, err := payingModel.CreatePaying("2027-02", "user:2", true)
	assert.NoError(err)
	manuallyClaimRunning(s.T(), failing.ID)
	assert.NoError(payingModel.FailPaying(failing, "synthetic"))
	reloaded, err = payingModel.GetPaying(failing.ID)
	assert.NoError(err)
	assert.Nil(reloaded.ActiveFlag, "failed must yield active_flag=NULL")
}

// TestActiveFlagUniquenessAcrossStatuses verifies that the unique key
// uk_active_paying(month, active_flag) enforces "at most one active paying
// per month" using the GENERATED active_flag values, not what the
// application might have written. The two-month test matrix:
//
//	same-month, status pending + pending           -> second insert fails (active_flag=1+1)
//	same-month, status pending + completed         -> second insert succeeds (active_flag=1 + NULL)
//	different-month, status pending + pending     -> second insert succeeds (1+1 but different month)
func (s *PayingSuite) TestActiveFlagUniquenessAcrossStatuses() {
	assert := s.Assert()

	// Same month, both pending: second must be rejected by
	// uk_active_paying (both rows have active_flag=1).
	first, err := payingModel.CreatePaying("2027-03", "user:1", false)
	assert.NoError(err)
	_, err = payingModel.CreatePaying("2027-03", "user:2", false)
	assert.Error(err)
	assert.True(isPayingAlreadyActive(err))

	// After completing `first`, a fresh pending for the same month
	// is allowed (active_flag NULL does not collide with 1 in MySQL).
	manuallyClaimRunning(s.T(), first.ID)
	assert.NoError(payingModel.CompletePaying(first, `{"ok":true}`))
	second, err := payingModel.CreatePaying("2027-03", "user:3", false)
	assert.NoError(err)
	assert.NotNil(second)

	// Different month: two pending rows coexist trivially.
	third, err := payingModel.CreatePaying("2027-04", "user:4", false)
	assert.NoError(err)
	fourth, err := payingModel.CreatePaying("2027-05", "user:5", false)
	assert.NoError(err)
	assert.NotNil(third)
	assert.NotNil(fourth)
}

// isPayingAlreadyActive mirrors the API-layer errors.Is check so the test
// does not depend on billing_errors.BaseError.Is semantics.
func isPayingAlreadyActive(err error) bool {
	if err == nil {
		return false
	}
	return errors.Is(err, billing_errors.PayingAlreadyActiveError)
}

func isObjectNotExists(err error) bool {
	return errors.Is(err, billing_errors.ObjectNotExistsError)
}

func isRecordNotFound(err error) bool {
	return err != nil && errors.Is(err, gorm.ErrRecordNotFound)
}

func uintPtr(v uint) *uint {
	return &v
}

// manuallyClaimRunning transitions a paying row from pending to running
// without going through ClaimNextPendingPaying. CompletePaying and FailPaying
// gate their UPDATE on status='running' (M1), so tests that drive the
// lifecycle directly must first claim the row at the DB level.
func manuallyClaimRunning(t *testing.T, payingID uint) {
	t.Helper()
	now := timeutils.GetNow()
	err := db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", payingID).
		Updates(map[string]interface{}{
			"status":       payment.PayingStatusRunning,
			"picked_up_at": &now,
		}).Error
	if err != nil {
		t.Fatalf("manuallyClaimRunning(id=%d): %v", payingID, err)
	}
}

// TestResetToPendingHappyPath verifies the transient-retry escape hatch:
// a paying currently in `running` state can be rewound to `pending` so the
// next worker tick re-claims it. The active_flag STORED GENERATED column
// must re-derive to 1 after status flips back to pending.
func (s *PayingSuite) TestResetToPendingHappyPath() {
	assert := s.Assert()
	paying, err := payingModel.CreatePaying("2027-05", "user:1", false)
	assert.NoError(err)

	// Walk pending -> running like ClaimNextPendingPaying would.
	now := timeutils.GetNow()
	assert.NoError(db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", paying.ID).
		Updates(map[string]interface{}{
			"status":       payment.PayingStatusRunning,
			"picked_up_at": now,
		}).Error)

	// Sanity: round-trip the running row so the in-memory struct mirrors DB.
	running, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(payment.PayingStatusRunning, running.Status)
	assert.NotNil(running.PickedUpAt)
	assert.NotNil(running.ActiveFlag)

	assert.NoError(payingModel.ResetToPending(running))

	// The in-memory struct reflects the reset.
	assert.Equal(payment.PayingStatusPending, running.Status)
	assert.Nil(running.PickedUpAt)
	assert.NotNil(running.ActiveFlag, "active_flag should be re-derived to 1 by the GENERATED column on next read")
	assert.Equal(uint(1), *running.ActiveFlag)

	// The DB row matches.
	reloaded, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(payment.PayingStatusPending, reloaded.Status)
	assert.Nil(reloaded.PickedUpAt)
	assert.NotNil(reloaded.ActiveFlag)
	assert.Equal(uint(1), *reloaded.ActiveFlag)
}

// TestResetToPendingRejectsTerminalStates guards against accidentally
// rewinding a completed/failed paying. The UPDATE's WHERE clause is gated
// on status='running' so any non-running row is a no-op + error.
func (s *PayingSuite) TestResetToPendingRejectsTerminalStates() {
	assert := s.Assert()
	paying, err := payingModel.CreatePaying("2027-06", "user:1", false)
	assert.NoError(err)

	// Walk pending -> running first; CompletePaying is gated on running.
	manuallyClaimRunning(s.T(), paying.ID)

	// Complete the paying first.
	assert.NoError(payingModel.CompletePaying(paying, `{"ok":true}`))
	completed, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(payment.PayingStatusCompleted, completed.Status)

	err = payingModel.ResetToPending(completed)
	assert.Error(err)
	assert.True(errors.Is(err, billing_errors.ObjectNotExistsError),
		"ResetToPending must refuse to rewind a completed paying, got %v", err)

	reloaded, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(payment.PayingStatusCompleted, reloaded.Status,
		"a completed paying must NOT be rewound to pending")
}

// TestHeartbeatTouchesRunningRow verifies Heartbeat bumps updated_at on a
// status='running' row and is a no-op once the row flips to a terminal
// status (so the janitor's "stuck since updated_at" signal never overrides
// the worker's terminal timestamp). It also asserts the alive bool
// invariant: true while the row is running, false once it has flipped to a
// terminal status — so the worker can bail out before writing
// CompletePaying/FailPaying.
func (s *PayingSuite) TestHeartbeatTouchesRunningRow() {
	assert := s.Assert()
	paying, err := payingModel.CreatePaying("2027-07", "user:hb", false)
	assert.NoError(err)

	// Move into running by hand to set up the precondition; claim would
	// also work but asserts the same shape.
	now := timeutils.GetNow()
	assert.NoError(db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", paying.ID).
		Updates(map[string]interface{}{
			"status":       payment.PayingStatusRunning,
			"picked_up_at": &now,
		}).Error)

	// First heartbeat should bump updated_at and report alive == true.
	before, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	updatedBefore := before.UpdatedAt

	time.Sleep(10 * time.Millisecond) // ensure clock advances past 1ms
	alive, err := payingModel.Heartbeat(paying.ID)
	assert.NoError(err)
	assert.True(alive, "heartbeat on a running row must report alive=true")
	after, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.True(after.UpdatedAt.After(updatedBefore) || after.UpdatedAt.Equal(updatedBefore),
		"heartbeat must not regress updated_at; before=%s after=%s",
		updatedBefore.Format(time.RFC3339Nano), after.UpdatedAt.Format(time.RFC3339Nano))

	// Complete the row; subsequent heartbeats must be a no-op so we never
	// overwrite the terminal updated_at that the janitor / dashboard read.
	assert.NoError(payingModel.CompletePaying(after, `{"ok":true}`))
	completed, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	updatedAtCompletion := completed.UpdatedAt

	time.Sleep(10 * time.Millisecond)
	alive, err = payingModel.Heartbeat(paying.ID)
	assert.NoError(err)
	assert.False(alive, "heartbeat on a terminal row must report alive=false")
	final, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(updatedAtCompletion.UnixNano(), final.UpdatedAt.UnixNano(),
		"heartbeat on a terminal row must NOT update updated_at")
}

// TestFailPayingRejectsTerminalStates guards against accidentally re-failing
// a paying already in a terminal state. The UPDATE's WHERE clause is gated
// on status='running' (matching the symmetric guard on CompletePaying), so
// any non-running row is a no-op + ObjectNotExistsError. The in-memory
// struct must not be mutated when RowsAffected == 0.
func (s *PayingSuite) TestFailPayingRejectsTerminalStates() {
	assert := s.Assert()
	paying, err := payingModel.CreatePaying("2027-11", "user:fail-twice", false)
	assert.NoError(err)
	// Walk pending -> running first; FailPaying is gated on running.
	manuallyClaimRunning(s.T(), paying.ID)

	// First FailPaying succeeds and writes the error_message.
	assert.NoError(payingModel.FailPaying(paying, "first failure"))
	first, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(payment.PayingStatusFailed, first.Status)
	assert.Equal("first failure", first.ErrorMessage)

	// Re-failing a failed row must fail with ObjectNotExistsError, and
	// must not overwrite the original error_message.
	err = payingModel.FailPaying(first, "second failure")
	assert.Error(err)
	assert.True(isObjectNotExists(err),
		"FailPaying must refuse to re-fail a terminal row, got %v", err)
	reloaded, err := payingModel.GetPaying(paying.ID)
	assert.NoError(err)
	assert.Equal(payment.PayingStatusFailed, reloaded.Status)
	assert.Equal("first failure", reloaded.ErrorMessage,
		"a failed paying's error_message must NOT be overwritten by a second fail attempt")

	// Symmetric check: a completed row must also be rejected.
	completed, err := payingModel.CreatePaying("2027-12", "user:complete-then-fail", false)
	assert.NoError(err)
	manuallyClaimRunning(s.T(), completed.ID)
	assert.NoError(payingModel.CompletePaying(completed, `{"ok":true}`))
	failed, err := payingModel.GetPaying(completed.ID)
	assert.NoError(err)
	err = payingModel.FailPaying(failed, "post-completion failure")
	assert.Error(err)
	assert.True(isObjectNotExists(err),
		"FailPaying must reject a completed row, got %v", err)
}

// TestListStuckRunning returns running payings older than the threshold and
// ignores pending / completed / failed rows.
func (s *PayingSuite) TestListStuckRunning() {
	assert := s.Assert()
	oldRunning, err := payingModel.CreatePaying("2027-08", "user:stuck", false)
	assert.NoError(err)
	freshRunning, err := payingModel.CreatePaying("2027-09", "user:alive", false)
	assert.NoError(err)
	pending, err := payingModel.CreatePaying("2027-10", "user:pending", false)
	assert.NoError(err)

	now := timeutils.GetNow()
	// oldRunning: 30 minutes ago, status=running → should appear in stuck list.
	oldTime := now.Add(-30 * time.Minute)
	assert.NoError(db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", oldRunning.ID).
		Updates(map[string]interface{}{
			"status":       payment.PayingStatusRunning,
			"picked_up_at": &oldTime,
			"updated_at":   oldTime,
		}).Error)
	// freshRunning: 1 minute ago, status=running → NOT stuck.
	freshTime := now.Add(-1 * time.Minute)
	assert.NoError(db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", freshRunning.ID).
		Updates(map[string]interface{}{
			"status":       payment.PayingStatusRunning,
			"picked_up_at": &freshTime,
			"updated_at":   freshTime,
		}).Error)

	stuck, err := payingModel.ListStuckRunning(10 * time.Minute)
	assert.NoError(err)

	ids := make([]uint, 0, len(stuck))
	for _, p := range stuck {
		ids = append(ids, p.ID)
	}
	assert.Contains(ids, oldRunning.ID, "30-minute-old running row must be flagged stuck")
	assert.NotContains(ids, freshRunning.ID, "1-minute-old running row is healthy")
	assert.NotContains(ids, pending.ID, "pending rows are not the janitor's concern")

	// Make sure FailPaying on a stuck row reaches the right terminal state —
	// this is the contract the worker relies on after we reworked the
	// CompletePaying error path to call FailPaying.
	assert.NoError(payingModel.FailPaying(oldRunning, "stuck-running regression test"))
	final, err := payingModel.GetPaying(oldRunning.ID)
	assert.NoError(err)
	assert.Equal(payment.PayingStatusFailed, final.Status)
	assert.Equal("stuck-running regression test", final.ErrorMessage)
}