package sales

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/TaruLab/keluargamart-backend/internal/db"
	"github.com/TaruLab/keluargamart-backend/internal/finance"
	"github.com/go-chi/chi/v5"
	"github.com/shopspring/decimal"
)

// The sweeper's guards, in the order they run. Each of these exists because the
// opposite is a way to lose money or double-book it, and every one of them is
// pinned here against the specific outcome it produces.

// sweepFixture is a 2x100 e-commerce order with an open invoice carrying a snap
// ref, selected as the sweep's only candidate, with the provider stub wired.
type sweepFixture struct {
	s      *Service
	fs     *fakeStore
	order  *SalesOrder
	inv    *Invoice
	ref    string
	status *statusStubRT
	// txBase is how many transactions the FIXTURE began (issuing the Snap
	// token takes one), so txs() returns only the sweep's own and a
	// commit/rollback assertion is about the sweep rather than about setup.
	txBase int
}

// txs is every transaction the SWEEP began.
func (f *sweepFixture) txs() []*fakeTx { return f.fs.txes[f.txBase:] }

func newSweepFixture(t *testing.T) *sweepFixture {
	t.Helper()
	s := svc()
	o, inv, ref := notifOrderWithRef(t, s)
	fs := s.store.(*fakeStore)
	fs.seedSweepCandidates(inv.ID)
	// The provider says "captured" by default: the settle path is the one most
	// of these tests exercise, and a fixture that made every case state its own
	// provider answer would let a guard be forgotten silently.
	st := statusSvc(t, s, "200.00")
	return &sweepFixture{s: s, fs: fs, order: o, inv: inv, ref: ref, status: st, txBase: len(fs.txes)}
}

func (f *sweepFixture) sweep(t *testing.T) *SnapSweepReport {
	t.Helper()
	// Only the sweep's own locked reads are attributed: the fixture got its
	// Snap token issued, which performed an order→invoice lock sequence of its
	// own, and that traffic is not what this suite is asserting about.
	f.fs.lockScopeStart("sweep")
	defer f.fs.lockScopeEnd()
	rep, err := f.s.SweepSnapInvoices(context.Background(), 0)
	if err != nil {
		t.Fatalf("sweep err = %v", err)
	}
	return rep
}

// An invoice somebody else already closed is COUNTED, never re-settled, and the
// provider is never asked about it: there is no open balance left to reconcile
// and a second probe would only risk a second payment row.
func TestSweep_AlreadySettledCountedAndNeverResettled(t *testing.T) {
	f := newSweepFixture(t)
	if _, err := f.fs.MarkInvoicePaid(context.Background(), nil, f.inv.ID, f.inv.TotalAmount); err != nil {
		t.Fatal(err)
	}
	before := len(f.fs.payments)
	orderBefore := orderPaid(t, f.s, f.order.ID)

	rep := f.sweep(t)

	if rep.AlreadySettled != 1 {
		t.Errorf("already_settled = %d, want 1", rep.AlreadySettled)
	}
	if rep.Settled != 0 {
		t.Errorf("settled = %d, want 0 (an already-settled invoice must not be settled again)", rep.Settled)
	}
	if f.status.calls != 0 {
		t.Errorf("status calls = %d, want 0 (there is nothing left to ask about)", f.status.calls)
	}
	if len(f.fs.payments) != before {
		t.Errorf("payment rows = %d, want %d (no second collection of the same money)", len(f.fs.payments), before)
	}
	if got := orderPaid(t, f.s, f.order.ID); !got.Equal(orderBefore) {
		t.Errorf("order paid_amount = %s, want %s (unchanged)", got, orderBefore)
	}
	if rep.StrandedOldest != nil {
		t.Errorf("stranded_oldest = %v, want nil (nothing is stranded)", *rep.StrandedOldest)
	}
}

// THE WRITTEN-OFF GUARD. CancelOrderInvoices leaves total_amount AND
// snap_order_ref in place, so a capture that lands after the order was
// cancelled reaches the sweeper looking exactly like a settlement. The sweeper
// must NOT resurrect it — the debt was deliberately forgiven — and it must
// report a COUNT, never an error, because an error is a redelivery loop against
// a provider that has nothing further to tell us.
func TestSweep_WrittenOffInvoiceIsCountedNotResurrected(t *testing.T) {
	f := newSweepFixture(t)
	cancelInvoices(t, f.s, f.order.ID)
	orderBefore := orderPaid(t, f.s, f.order.ID)

	rep, err := f.s.SweepSnapInvoices(context.Background(), 0)
	if err != nil {
		t.Fatalf("a written-off capture must not fail the pass: %v", err)
	}

	if rep.WrittenOff != 1 {
		t.Errorf("written_off = %d, want 1", rep.WrittenOff)
	}
	if rep.Settled != 0 {
		t.Errorf("settled = %d, want 0 (a forgiven debt must stay forgiven)", rep.Settled)
	}
	if rep.HardErrors != 0 {
		t.Errorf("hard_errors = %d, want 0 (a written-off capture is a count, not an error)", rep.HardErrors)
	}
	// The counter is written_off, NOT already_settled. The two share a code path
	// shape — both return without settling — so without this assertion a
	// regression that reported the capture as a benign no-op instead of the
	// alarm it is would still pass everything above.
	if rep.AlreadySettled != 0 {
		t.Errorf("already_settled = %d, want 0: a written-off capture is not an already-settled invoice", rep.AlreadySettled)
	}
	if f.inv.Status != "cancelled" {
		t.Errorf("invoice status = %q, want cancelled (not resurrected)", f.inv.Status)
	}
	if f.status.calls != 0 {
		t.Errorf("status calls = %d, want 0 (the guard runs BEFORE the provider is contacted)", f.status.calls)
	}
	noPaymentFor(t, f.fs, f.inv.ID)
	if got := orderPaid(t, f.s, f.order.ID); !got.Equal(orderBefore) {
		t.Errorf("order paid_amount = %s, want %s (nothing settled, nothing moved)", got, orderBefore)
	}
}

// THE ZERO-ROW BACKSTOP. MarkInvoicePaid's WHERE clause is
// `status IN ('unpaid','partial')`, so 0 rows means a concurrent caller closed
// the invoice between this transaction's locked read and this write. The sweeper
// must roll back and count it as already-settled. Creating the payment row
// anyway would book a second collection of the same money, and this is the
// guard that makes a THIRD concurrent caller safe by construction rather than by
// convention.
func TestSweep_ZeroRowMarkPaidRollsBackAndBooksNoPayment(t *testing.T) {
	f := newSweepFixture(t)
	f.fs.markPaidZeroRows = true

	rep := f.sweep(t)

	if rep.AlreadySettled != 1 {
		t.Errorf("already_settled = %d, want 1 (someone else won the row)", rep.AlreadySettled)
	}
	if rep.Settled != 0 {
		t.Errorf("settled = %d, want 0", rep.Settled)
	}
	noPaymentFor(t, f.fs, f.inv.ID)
	txs := f.txs()
	if len(txs) == 0 {
		t.Fatal("no transaction was begun")
	}
	if txs[0].commits != 0 {
		t.Errorf("commits = %d, want 0: the backstop must ROLL BACK, not commit a partial settle", txs[0].commits)
	}
	if txs[0].rollbacks == 0 {
		t.Error("no rollback: the transaction that lost the race was left open")
	}
}

// Amount reconciliation runs BEFORE any write and fails closed. A gross we
// cannot account for must leave no trace: no payment row, no paid invoice, no
// order movement — and the invoice open, for a human or a corrected provider
// figure to look at.
func TestSweep_AmountMismatchFailsClosed(t *testing.T) {
	for _, tc := range []struct{ name, gross string }{
		{"under-payment", "150.00"},
		{"over-payment", "999.00"},
		{"zero", "0.00"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSweepFixture(t)
			statusSvc(t, f.s, tc.gross)
			orderBefore := orderPaid(t, f.s, f.order.ID)

			rep := f.sweep(t)

			if rep.AmountMismatch != 1 {
				t.Errorf("amount_mismatch = %d, want 1", rep.AmountMismatch)
			}
			if rep.Settled != 0 {
				t.Errorf("settled = %d, want 0 (a mismatch must never settle)", rep.Settled)
			}
			if f.inv.Status == "paid" {
				t.Error("invoice marked paid on an amount mismatch")
			}
			noPaymentFor(t, f.fs, f.inv.ID)
			if got := orderPaid(t, f.s, f.order.ID).Sub(orderBefore); !got.IsZero() {
				t.Errorf("order advanced by %s, want 0", got)
			}
			// A mismatch is not stamped, so the next pass looks again.
			if len(f.fs.snapChecked) != 0 {
				t.Errorf("snap_checked = %v, want none (a refused amount must be retried, not backed off)", f.fs.snapChecked)
			}
		})
	}
}

// THE STAMP BOUNDARY, both directions, in one test because the pair IS the
// contract:
//
//   - a hard provider error rolls the stamp back, so the invoice is probed again
//     on the next pass (a transient failure must never be cached as "checked");
//   - a soft no-settlement COMMITS the stamp, so an abandoned invoice backs off
//     for the throttle window instead of being re-probed every 15 minutes.
//
// Both come from the same transaction boundary. A stamp written on its own
// transaction, or on an error path, would silently break one of them.
func TestSweep_HardErrorRollsBackStampSoftNoSettlementCommitsIt(t *testing.T) {
	t.Run("hard-error", func(t *testing.T) {
		f := newSweepFixture(t)
		// A transport failure below HTTP: a dead gateway, not an answer. A
		// timeout is classified the same way and for the same reason.
		f.s.WithHTTPClient(&http.Client{Transport: &statusStubRT{byRef: map[string]statusReply{
			f.ref: {transportErr: errors.New("dial tcp: connection refused")},
		}}})

		rep := f.sweep(t)

		if rep.HardErrors != 1 {
			t.Errorf("hard_errors = %d, want 1", rep.HardErrors)
		}
		if rep.Settled != 0 || rep.Pending != 0 || rep.UnknownRef != 0 {
			t.Errorf("a hard error booked an outcome: settled %d pending %d unknown %d, want all 0",
				rep.Settled, rep.Pending, rep.UnknownRef)
		}
		if len(f.fs.snapChecked) != 0 {
			t.Errorf("snap_checked = %v, want none: a hard error must leave the invoice retryable", f.fs.snapChecked)
		}
		txs := f.txs()
		if len(txs) == 0 {
			t.Fatal("no transaction was begun")
		}
		if txs[0].commits != 0 {
			t.Errorf("commits = %d, want 0 (the failed probe must not commit)", txs[0].commits)
		}
		if txs[0].rollbacks == 0 {
			t.Error("no rollback on the hard-error path")
		}
		if rep.StrandedOldest == nil || *rep.StrandedOldest != f.inv.InvoiceNumber {
			t.Errorf("stranded_oldest = %v, want %q (a failed probe leaves the invoice open)",
				derefStr(rep.StrandedOldest), f.inv.InvoiceNumber)
		}
	})

	t.Run("soft-no-settlement", func(t *testing.T) {
		f := newSweepFixture(t)
		// The provider holds a transaction that has not settled: the buyer
		// genuinely has not finished paying.
		statusSvc(t, f.s, "200.00").status = "pending"

		rep := f.sweep(t)

		if rep.Pending != 1 {
			t.Errorf("pending = %d, want 1", rep.Pending)
		}
		if len(f.fs.snapChecked) != 1 || f.fs.snapChecked[0] != f.inv.ID {
			t.Fatalf("snap_checked = %v, want the invoice stamped once so it backs off", f.fs.snapChecked)
		}
		// The stamp must have ridden the caller's transaction, not one of its own.
		txs := f.txs()
		if len(txs) == 0 {
			t.Fatal("no transaction was begun")
		}
		if len(f.fs.snapCheckedQ) != 1 || f.fs.snapCheckedQ[0] != txs[0] {
			t.Errorf("stamp arrived on a different transaction than the probe: %v vs %v",
				f.fs.snapCheckedQ, txs[0])
		}
		if txs[0].commits != 1 {
			t.Errorf("commits = %d, want 1 (a completed probe backs off)", txs[0].commits)
		}
	})
}

// The two SOFT outcomes are different facts and get different counters. A ref
// the provider has never seen means the buyer abandoned Snap: final for that
// ref, and re-asking will not change it. A transaction the provider holds but
// has not settled means the buyer is mid-payment: still open, and the invoice
// backs off rather than being re-probed every pass.
func TestSweep_DistinguishesUnknownRefFromPending(t *testing.T) {
	t.Run("unknown-ref", func(t *testing.T) {
		f := newSweepFixture(t)
		// Neither the stored ref nor the canonical one is known to the provider.
		f.s.WithHTTPClient(&http.Client{Transport: &statusStubRT{byRef: map[string]statusReply{
			f.ref: {notFound: true},
		}}})
		rep := f.sweep(t)
		if rep.UnknownRef != 1 {
			t.Errorf("unknown_ref = %d, want 1", rep.UnknownRef)
		}
		if rep.Pending != 0 {
			t.Errorf("pending = %d, want 0", rep.Pending)
		}
		if len(f.fs.snapChecked) != 1 {
			t.Errorf("snap_checked = %v, want one stamp (a completed probe backs off either way)", f.fs.snapChecked)
		}
	})

	t.Run("pending", func(t *testing.T) {
		f := newSweepFixture(t)
		statusSvc(t, f.s, "200.00").status = "pending"
		rep := f.sweep(t)
		if rep.Pending != 1 {
			t.Errorf("pending = %d, want 1", rep.Pending)
		}
		if rep.UnknownRef != 0 {
			t.Errorf("unknown_ref = %d, want 0", rep.UnknownRef)
		}
	})
}

// An invoice Midtrans was never asked to charge has no ref to probe. Counted,
// never an error, and never stamped — there is no probe result to record.
func TestSweep_NoRefIsNotAttempted(t *testing.T) {
	f := newSweepFixture(t)
	f.fs.invoices[f.inv.ID].SnapOrderRef = nil
	f.s.WithHTTPClient(&http.Client{Transport: &statusStubRT{byRef: map[string]statusReply{}}})

	rep := f.sweep(t)

	if rep.NotAttempted != 1 {
		t.Errorf("not_attempted = %d, want 1", rep.NotAttempted)
	}
	if f.status.calls != 0 {
		t.Errorf("status calls = %d, want 0 (no ref, no probe)", f.status.calls)
	}
	if len(f.fs.snapChecked) != 0 {
		t.Errorf("snap_checked = %v, want none (no probe ran)", f.fs.snapChecked)
	}
}

// THE REGRESSION, on the sweeper. reconcileSnapGross returns the figure the
// payment row must record, and it is the ONLY definition of it. The raw
// inv.RemainingAmount column is identical to that figure whenever the invoice
// carries an open balance, so a third copy of this bug would ship invisibly
// again. With a NON-POSITIVE open balance the two differ, and the old code
// booked zero (or a negative) while advancing the order by it and flipping the
// invoice to paid: a completed sale with no money recorded for it.
func TestSweep_BooksReconciledPayableNotRemaining(t *testing.T) {
	for _, tc := range []struct {
		name      string
		remaining decimal.Decimal
		want      decimal.Decimal
	}{
		// The bug's shape. payable falls back to the invoice total; the raw
		// column is 0 and -0.50.
		{"zero-remaining", decimal.Zero, fracTotal},
		{"negative-remaining", decimal.RequireFromString("-0.50"), fracTotal},
		// The unchanged normal case: the open balance IS the payable.
		{"positive-remaining", decimal.RequireFromString("120.25"), decimal.RequireFromString("120.25")},
	} {
		t.Run(tc.name, func(t *testing.T) {
			s := svc()
			_, inv, _, _ := fracOrderWithRef(t, s)
			fs := s.store.(*fakeStore)
			fs.invoices[inv.ID].RemainingAmount = tc.remaining
			fs.seedSweepCandidates(inv.ID)
			o := fs.orders[inv.SalesOrderID]
			before := orderPaid(t, s, o.ID)
			// The provider echoes the ROUNDED amount Snap was told for this
			// payable, so the amount check passes and only the booking can differ.
			statusSvc(t, s, snapRoundedGross(tc.want).StringFixed(2))

			rep, err := s.SweepSnapInvoices(context.Background(), 0)
			if err != nil {
				t.Fatalf("sweep err = %v, want a settle booked at %s", err, tc.want)
			}
			if rep.Settled != 1 {
				t.Fatalf("settled = %d, want 1 (all counters: %+v)", rep.Settled, rep)
			}
			if got := payAmountFor(t, fs, inv.ID); !got.Equal(tc.want) {
				t.Errorf("payment amount = %s, want %s (the reconciled payable, not remaining_amount %s)",
					got, tc.want, tc.remaining)
			}
			if got := orderPaid(t, s, o.ID).Sub(before); !got.Equal(tc.want) {
				t.Errorf("order advanced by %s, want %s (the same figure the payment row recorded)", got, tc.want)
			}
		})
	}
}

// THREE-WAY PARITY on one invoice shape. The webhook, the buyer's poll and the
// sweeper are the same money path reached by three doors; a ledger that records
// a different figure depending on which door the payment came through is a
// ledger nobody can reconcile. The figure they all record is the one
// reconcileSnapGross RETURNED.
func TestSweep_MatchesWebhookAndPollPaymentAmount(t *testing.T) {
	for _, tc := range []struct {
		name      string
		remaining decimal.Decimal
	}{
		{"positive-remaining", fracTotal},
		{"zero-remaining", decimal.Zero},
		{"negative-remaining", decimal.RequireFromString("-0.50")},
	} {
		t.Run(tc.name, func(t *testing.T) {
			// Webhook.
			w := svc()
			_, winv, ref, _ := fracOrderWithRef(t, w)
			wfs := w.store.(*fakeStore)
			wfs.invoices[winv.ID].RemainingAmount = tc.remaining
			wantPayable := invoicePayableAmount(wfs.invoices[winv.ID])
			if out, err := w.HandleMidtransNotification(context.Background(), notifFor(ref, "321.00")); err != nil || out != finance.MidtransOutcomeSettled {
				t.Fatalf("webhook outcome = %q err = %v, want settled", out, err)
			}
			viaWebhook := payAmountFor(t, wfs, winv.ID)

			// Poll.
			p := svc()
			po, pinv, _, _ := fracOrderWithRef(t, p)
			pfs := p.store.(*fakeStore)
			pfs.invoices[pinv.ID].RemainingAmount = tc.remaining
			statusSvc(t, p, "321.00")
			if out, err := p.SnapPaymentStatus(context.Background(), po.ID, 42); err != nil || out.PaymentStatus != "settled" {
				t.Fatalf("poll result = %+v err = %v, want settled", out, err)
			}
			viaPoll := payAmountFor(t, pfs, pinv.ID)

			// Sweeper, on an identical fixture.
			s := svc()
			_, sinv, _, _ := fracOrderWithRef(t, s)
			sfs := s.store.(*fakeStore)
			sfs.invoices[sinv.ID].RemainingAmount = tc.remaining
			sfs.seedSweepCandidates(sinv.ID)
			so := sfs.orders[sinv.SalesOrderID]
			sweepBefore := orderPaid(t, s, so.ID)
			statusSvc(t, s, "321.00")
			rep, err := s.SweepSnapInvoices(context.Background(), 0)
			if err != nil || rep.Settled != 1 {
				t.Fatalf("sweep settled = %d err = %v, want 1 settled", rep.Settled, err)
			}
			viaSweep := payAmountFor(t, sfs, sinv.ID)
			sweepAdvance := orderPaid(t, s, so.ID).Sub(sweepBefore)

			if !viaWebhook.Equal(wantPayable) {
				t.Fatalf("fixture: webhook booked %s, want the payable %s", viaWebhook, wantPayable)
			}
			if !viaPoll.Equal(viaWebhook) {
				t.Errorf("poll booked %s, webhook booked %s", viaPoll, viaWebhook)
			}
			if !viaSweep.Equal(viaWebhook) {
				t.Errorf("sweeper booked %s, webhook booked %s — all three settle paths must record one figure",
					viaSweep, viaWebhook)
			}
			if !sweepAdvance.Equal(viaSweep) {
				t.Errorf("sweeper advanced the order by %s but booked %s", sweepAdvance, viaSweep)
			}
		})
	}
}

// THE LOCK ORDER, as a sequence. Every other path in this package that settles
// an invoice takes the sales_orders row FOR UPDATE and only then the invoice
// row. The reverse order here would be a textbook ABBA cycle against
// CancelOrderInvoices (order lock, then the invoice row): it works for months,
// passes every functional test, and then deadlocks in production during a
// cancel. A set-shaped assertion could not tell the two orders apart, so the
// sequence is what is pinned.
func TestSweep_LocksOrderBeforeInvoice(t *testing.T) {
	f := newSweepFixture(t)
	f.sweep(t)

	seq := f.fs.lockSeq()
	if len(seq) < 2 {
		t.Fatalf("locked reads = %v, want at least the order then the invoice", seq)
	}
	if seq[0] != "sweep:order" || seq[1] != "sweep:invoice" {
		t.Errorf("locked reads = %v, want the sales_orders row before the invoice row", seq)
	}
	// InvoiceForOrder, NOT OpenInvoiceForOrder: the status predicate on the
	// latter would hide a 'paid' invoice as ErrNotFound, so an already-settled
	// invoice the pass should have COUNTED would look like a missing row.
	for _, k := range seq {
		if k == "sweep:invoice-open" {
			t.Errorf("locked reads = %v: the sweeper must use InvoiceForOrder, not OpenInvoiceForOrder", seq)
			break
		}
	}
}

// A pass reports one outcome per candidate it reached, and the counters sum to
// the candidate set. A report whose numbers do not add up cannot be read by an
// operator at 3am, and it is the only structural check that every outcome enum
// value has a counter — a new outcome that record() forgets would silently
// vanish from the report.
//
// One store, three candidates, three different outcomes in one pass, so this
// covers more than one enum value at a time.
func TestSweep_ReportCountersAccountForEveryCandidate(t *testing.T) {
	f := newSweepFixture(t) // settled at the provider
	settledInv := f.inv.ID

	// A second, already-closed invoice on its own order.
	fs := f.fs
	fs.stocks["1:1"] = &Stock{ID: 1, ProductID: 1, WarehouseID: 1, Quantity: decimal.NewFromInt(10)}
	o2, err := f.s.PlaceCheckout(context.Background(), CheckoutInput{
		Items: []CheckoutItemInput{{ProductID: 1, Quantity: decimal.NewFromInt(2)}},
	}, 42, "Buyer2", "buyer2@keluargamart.local")
	if err != nil {
		t.Fatal(err)
	}
	paidInv, err := fs.CreateInvoice(context.Background(), nil, createInvoiceParams(&Invoice{
		InvoiceNumber: "INV-T-PAID", SalesOrderID: o2.ID, CustomerID: o2.CustomerID, Status: "unpaid",
		InvoiceDate: "2026-09-20", DueDate: "2026-09-20",
		TotalAmount: o2.TotalAmount, RemainingAmount: o2.TotalAmount,
	}))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fs.MarkInvoicePaid(context.Background(), nil, paidInv, o2.TotalAmount); err != nil {
		t.Fatal(err)
	}

	// A third, written off, whose ref survives the cancel — the case the
	// candidate SQL would exclude but a stale selection can still carry.
	o3, err := f.s.PlaceCheckout(context.Background(), CheckoutInput{
		Items: []CheckoutItemInput{{ProductID: 1, Quantity: decimal.NewFromInt(2)}},
	}, 42, "Buyer3", "buyer3@keluargamart.local")
	if err != nil {
		t.Fatal(err)
	}
	offInv, err := fs.CreateInvoice(context.Background(), nil, createInvoiceParams(&Invoice{
		InvoiceNumber: "INV-T-OFF", SalesOrderID: o3.ID, CustomerID: o3.CustomerID, Status: "unpaid",
		InvoiceDate: "2026-09-20", DueDate: "2026-09-20",
		TotalAmount: o3.TotalAmount, RemainingAmount: o3.TotalAmount,
	}))
	if err != nil {
		t.Fatal(err)
	}
	ref := "KM-INV-T-OFF-" + itoa2(o3.ID)
	fs.invoices[offInv].SnapOrderRef = &ref
	cancelInvoices(t, f.s, o3.ID)

	// Two stale selections, the two ways one happens: a candidate naming a row
	// that is gone entirely, and one naming a real order whose invoice has been
	// replaced by a different row.
	stale := SnapSweepCandidate{ID: 0, InvoiceNumber: "INV-T-GONE", SalesOrderID: 0}
	moved := SnapSweepCandidate{ID: 0, InvoiceNumber: "INV-T-WRONG", SalesOrderID: o2.ID}
	fs.sweepCandidates = []SnapSweepCandidate{
		{ID: settledInv, InvoiceNumber: fs.invoices[settledInv].InvoiceNumber, SalesOrderID: fs.invoices[settledInv].SalesOrderID},
		{ID: paidInv, InvoiceNumber: fs.invoices[paidInv].InvoiceNumber, SalesOrderID: fs.invoices[paidInv].SalesOrderID},
		{ID: offInv, InvoiceNumber: fs.invoices[offInv].InvoiceNumber, SalesOrderID: fs.invoices[offInv].SalesOrderID},
		stale, moved,
	}

	rep := f.sweep(t)

	if rep.Candidates != 5 {
		t.Fatalf("candidates = %d, want 5", rep.Candidates)
	}
	if rep.Settled != 1 || rep.AlreadySettled != 1 || rep.WrittenOff != 1 {
		t.Errorf("settled %d / already_settled %d / written_off %d, want 1/1/1 (all: %+v)",
			rep.Settled, rep.AlreadySettled, rep.WrittenOff, rep)
	}
	if rep.Skipped != 2 {
		t.Errorf("skipped = %d, want 2 (a vanished row, and a row that is no longer the order's invoice)",
			rep.Skipped)
	}
	if got := rep.Settled + rep.AlreadySettled + rep.WrittenOff + rep.NotAttempted + rep.Pending +
		rep.UnknownRef + rep.HardErrors + rep.AmountMismatch + rep.Skipped; got != rep.Candidates {
		t.Errorf("counters sum to %d, candidates = %d: every reached candidate must land in exactly one", got, rep.Candidates)
	}
	if rep.DurationMS < 0 {
		t.Errorf("duration_ms = %d, want >= 0", rep.DurationMS)
	}
}

// The sweeper never un-pays. A provider answer that is not a settlement must
// leave the invoice exactly as it found it, whatever the reason: there is
// deliberately no deny, cancel, expire or failure branch, and this asserts that
// absence rather than trusting it.
func TestSweep_NeverUnpaysOnANonSettlingProviderAnswer(t *testing.T) {
	for _, status := range []string{"pending", "deny", "cancel", "expire", "failure", "capture"} {
		t.Run(status, func(t *testing.T) {
			f := newSweepFixture(t)
			// capture + a fraud challenge maps to "ignored": the transaction is
			// held for review and must not settle.
			fraud := ""
			if status == "capture" {
				fraud = "challenge"
			}
			st := statusSvc(t, f.s, "200.00")
			st.status, st.fraud = status, fraud
			orderBefore := orderPaid(t, f.s, f.order.ID)

			rep := f.sweep(t)

			if rep.Settled != 0 {
				t.Errorf("settled = %d on a %q answer, want 0", rep.Settled, status)
			}
			if f.inv.Status == "paid" {
				t.Errorf("invoice paid on a %q answer", status)
			}
			noPaymentFor(t, f.fs, f.inv.ID)
			if got := orderPaid(t, f.s, f.order.ID).Sub(orderBefore); !got.IsZero() {
				t.Errorf("order advanced by %s on a %q answer, want 0", got, status)
			}
		})
	}
}

// An unset sweep token is 404, NOT 503. A 503 on this route tells a scanner
// that a payment sweeper is deployed here and is currently off; a 404 is
// indistinguishable from a route this build never had.
func TestSweep_HandlerUnsetTokenIsNotFound(t *testing.T) {
	// A fully working sweep that would answer 200 if the gate were not there:
	// a candidate, a provider, a server key. So the status this asserts is the
	// gate's decision and nothing else.
	f := newSweepFixture(t)
	s := f.s.WithSnapSweepToken("").WithMidtrans(MidtransConfig{ServerKey: "test-server-key"})
	h := NewHandler(s)
	rec := httptest.NewRecorder()
	h.SweepSnapInvoices(rec, httptest.NewRequest(http.MethodPost, "/api/v1/midtrans/sweep", nil))

	if rec.Code != http.StatusNotFound {
		t.Errorf("status = %d, want 404 (a 503 would advertise a deployed sweeper)", rec.Code)
	}
	if strings.Contains(rec.Body.String(), "sweep") {
		t.Errorf("body mentions the sweep: %s", rec.Body.String())
	}
}

// The token gate is the route's only authentication — crontab holds no user
// session — so a wrong or missing token must be rejected, and a correct one
// must run the pass. The suite is mounted the way main.go mounts it, gate
// included, so the route cannot be registered without its gate and still pass
// here.
func TestSweep_HandlerTokenGate(t *testing.T) {
	for _, tc := range []struct {
		name, auth string
		want       int
	}{
		{"correct", "Bearer correct-horse-battery-staple", http.StatusOK},
		{"wrong", "Bearer wrong", http.StatusUnauthorized},
		{"absent", "", http.StatusUnauthorized},
		// A prefix of the real token must not pass: the compare is over
		// fixed-length digests, so a partial guess is simply a different value.
		{"prefix", "Bearer correct-horse", http.StatusUnauthorized},
		// A token with the right prefix and extra characters must not pass
		// either — the length-difference case ConstantTimeCompare short-circuits.
		{"extended", "Bearer correct-horse-battery-staple-extra", http.StatusUnauthorized},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSweepFixture(t)
			f.s.WithSnapSweepToken("correct-horse-battery-staple")
			h := NewHandler(f.s)

			mux := chi.NewRouter()
			mux.Route("/api/v1/midtrans", func(mr chi.Router) {
				mr.Use(h.SnapSweepTokenGate)
				mr.Post("/sweep", h.SweepSnapInvoices)
			})
			req := httptest.NewRequest(http.MethodPost, "/api/v1/midtrans/sweep", strings.NewReader(`{"limit":10}`))
			if tc.auth != "" {
				req.Header.Set("Authorization", tc.auth)
			}
			rec := httptest.NewRecorder()
			mux.ServeHTTP(rec, req)

			if rec.Code != tc.want {
				t.Errorf("status = %d, want %d: %s", rec.Code, tc.want, rec.Body.String())
			}
			if tc.want == http.StatusOK {
				var env struct {
					Data SnapSweepReport `json:"data"`
				}
				if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
					t.Fatalf("response body: %v (%s)", err, rec.Body.String())
				}
				if env.Data.Settled != 1 {
					t.Errorf("settled = %d, want 1", env.Data.Settled)
				}
			}
		})
	}
}

// The 404-when-unset gate is a ROUTE middleware covering the whole group, not
// only a branch inside the sweep handler. That is the property worth pinning,
// and it is pinned with a PROBE handler that shares no code with the sweep: the
// probe 404s purely because of the middleware, so a gate that existed only
// inside SweepSnapInvoices would leave the probe reachable and fail here. That
// is exactly the failure mode being guarded against — the next route added to
// the group inheriting the gate for free.
func TestSweep_TokenGateMiddlewareHidesTheRouteWhenUnset(t *testing.T) {
	f := newSweepFixture(t)
	f.s.WithSnapSweepToken("")
	h := NewHandler(f.s)
	probe := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusTeapot)
	})

	t.Run("token-unset-hides-every-route", func(t *testing.T) {
		mux := chi.NewRouter()
		mux.Route("/api/v1/midtrans", func(mr chi.Router) {
			mr.Use(h.SnapSweepTokenGate)
			mr.Post("/sweep", probe)
			mr.Post("/anything-else-added-later", probe)
		})
		// A correct-looking header: the 404 is the gate's decision, not the
		// handler's.
		req := httptest.NewRequest(http.MethodPost, "/api/v1/midtrans/sweep", nil)
		req.Header.Set("Authorization", "Bearer anything")
		rec := httptest.NewRecorder()
		mux.ServeHTTP(rec, req)
		if rec.Code != http.StatusNotFound {
			t.Errorf("status = %d, want 404 from the gate middleware", rec.Code)
		}
	})

	t.Run("token-set-passes-through", func(t *testing.T) {
		h.svc.WithSnapSweepToken("t")
		mux := chi.NewRouter()
		mux.Route("/api/v1/midtrans", func(mr chi.Router) {
			mr.Use(h.SnapSweepTokenGate)
			mr.Post("/sweep", probe)
		})
		rec := httptest.NewRecorder()
		mux.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/v1/midtrans/sweep", nil))
		// The gate is an EXISTENCE check only; the constant-time compare is the
		// handler's. So a request with no token reaches the handler, and the
		// probe answers. This asserts the gate does not over-reach into a second
		// comparison that would have to be kept in sync.
		if rec.Code != http.StatusTeapot {
			t.Errorf("status = %d, want the gate to pass through to the handler", rec.Code)
		}
	})
}

// The clamp is the only thing a caller can influence, and it is applied in the
// SERVICE, not only in the handler, so a second caller (a future in-process job)
// gets the same bound.
func TestSweep_ServiceClampsLimitBeforeSelecting(t *testing.T) {
	f := newSweepFixture(t)
	store := &recordingCandidateStore{fakeStore: f.fs}
	s := NewService(store, &fakeSeq{}, time.UTC)
	s.WithMidtrans(MidtransConfig{ServerKey: "test-server-key", CoreBaseURL: "http://example.test/v2"}).
		WithHTTPClient(&http.Client{Transport: f.status})

	if _, err := s.SweepSnapInvoices(context.Background(), 100_000); err != nil {
		t.Fatal(err)
	}
	if store.got != snapSweepBatchCap {
		t.Errorf("candidate query limit = %d, want the server clamp %d", store.got, snapSweepBatchCap)
	}
}

// recordingCandidateStore remembers the limit the service asked for.
type recordingCandidateStore struct {
	*fakeStore
	got int
}

func (s *recordingCandidateStore) ListSnapSweepCandidates(_ context.Context, limit int) ([]SnapSweepCandidate, error) {
	s.got = limit
	return s.fakeStore.ListSnapSweepCandidates(context.Background(), limit)
}

// The batch cap is the ONLY thing a caller can influence, and it is clamped
// server-side. A caller asking for more than the cap gets the cap, not an
// error and not an unbounded scan.
func TestSweep_ClampsCallerLimit(t *testing.T) {
	for _, tc := range []struct {
		name string
		in   int
		want int
	}{
		{"zero-takes-the-cap", 0, snapSweepBatchCap},
		{"negative-takes-the-cap", -5, snapSweepBatchCap},
		{"over-cap-is-cut", 10_000, snapSweepBatchCap},
		{"in-range-passes-through", 10, 10},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if got := clampSnapSweepLimit(tc.in); got != tc.want {
				t.Errorf("clampSnapSweepLimit(%d) = %d, want %d", tc.in, got, tc.want)
			}
		})
	}
}

// No Midtrans server key means no probe is possible. That is a 503 through the
// ErrGatewayDisabled convention, and it is a DIFFERENT fact from an unset sweep
// token (404) — one is our misconfiguration, the other means the sweeper is not
// deployed.
func TestSweep_GatewayDisabledIsAServiceError(t *testing.T) {
	s := svc() // no WithMidtrans
	if _, err := s.SweepSnapInvoices(context.Background(), 0); !errors.Is(err, ErrGatewayDisabled) {
		t.Errorf("err = %v, want ErrGatewayDisabled", err)
	}
}

// TWO CONCURRENT SWEEPS ON ONE INVOICE SETTLE IT ONCE. The row locks serialise
// the two passes, and the zero-row backstop refuses the loser. Together they
// are what makes a third caller safe; the assertion is that the money is
// collected once, not that the mechanism is spelled a particular way.
func TestSweep_ConcurrentPassesSettleOnce(t *testing.T) {
	f := newSweepFixture(t)
	// A barrier at the provider call: both passes have taken their locks and
	// both are about to learn the invoice settled, so the second one to reach
	// MarkInvoicePaid is genuinely racing the first.
	barrier := newStatusBarrier(2, &statusStubRT{status: "settlement"})
	store := &sweepRaceStore{fakeStore: f.fs}

	var wg sync.WaitGroup
	reports := make([]*SnapSweepReport, 2)
	errs := make([]error, 2)
	for i := range reports {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			s := NewService(store, &lockedSeq{}, time.UTC)
			s.WithMidtrans(MidtransConfig{ServerKey: "test-server-key", CoreBaseURL: "http://example.test/v2"}).
				WithHTTPClient(&http.Client{Transport: barrier})
			reports[n], errs[n] = s.SweepSnapInvoices(context.Background(), 0)
		}(i)
	}
	wg.Wait()

	settled, already := 0, 0
	for i, err := range errs {
		if err != nil {
			t.Fatalf("pass %d err = %v", i, err)
		}
		settled += reports[i].Settled
		already += reports[i].AlreadySettled
	}
	if settled != 1 {
		t.Errorf("settled across both passes = %d, want exactly 1 (the money is collected once)", settled)
	}
	if already != 1 {
		t.Errorf("already_settled across both passes = %d, want 1 (the loser takes the backstop)", already)
	}
	if len(f.fs.payments) != 1 {
		t.Errorf("payment rows = %d, want 1 (one collection of the money)", len(f.fs.payments))
	}
}

// sweepRaceStore serialises the store calls two concurrent passes make on one
// fake, standing in for what FOR UPDATE does in the database: without it the Go
// maps in the fake are raced by the test harness itself rather than by the
// sweeper, and the assertion below would be measuring the fake's maps.
//
// It overrides ONLY the calls the settle path makes on shared rows. Everything
// else is promoted from the embedded fake.
type sweepRaceStore struct {
	*fakeStore
	mu sync.Mutex
}

func (s *sweepRaceStore) Begin(ctx context.Context) (Tx, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.Begin(ctx)
}

func (s *sweepRaceStore) OrderByIDForUpdate(ctx context.Context, q db.DBTX, id int64) (db.GetSalesOrderByIDForUpdateRow, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.OrderByIDForUpdate(ctx, q, id)
}

func (s *sweepRaceStore) OrderByID(ctx context.Context, id int64) (db.GetSalesOrderByIDRow, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.OrderByID(ctx, id)
}

func (s *sweepRaceStore) InvoiceForOrder(ctx context.Context, q db.DBTX, orderID int64) (*Invoice, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.InvoiceForOrder(ctx, q, orderID)
}

func (s *sweepRaceStore) InvoiceBySnapRef(ctx context.Context, q db.DBTX, ref string) (*Invoice, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.InvoiceBySnapRef(ctx, q, ref)
}

func (s *sweepRaceStore) MarkInvoicePaid(ctx context.Context, q db.DBTX, id int64, paid decimal.Decimal) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.MarkInvoicePaid(ctx, q, id, paid)
}

func (s *sweepRaceStore) CreatePayment(ctx context.Context, q db.DBTX, p db.CreateSalesPaymentParams) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.CreatePayment(ctx, q, p)
}

func (s *sweepRaceStore) CreateGatewayPayment(ctx context.Context, q db.DBTX, p db.CreateSalesPaymentParams, gateway, txnID, status string) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.CreateGatewayPayment(ctx, q, p, gateway, txnID, status)
}

func (s *sweepRaceStore) SettleOrderPayment(ctx context.Context, q db.DBTX, orderID int64, amount decimal.Decimal) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.SettleOrderPayment(ctx, q, orderID, amount)
}

func (s *sweepRaceStore) SetInvoiceSnapChecked(ctx context.Context, q db.DBTX, id int64) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.SetInvoiceSnapChecked(ctx, q, id)
}

func (s *sweepRaceStore) ListSnapSweepCandidates(ctx context.Context, limit int) ([]SnapSweepCandidate, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.fakeStore.ListSnapSweepCandidates(ctx, limit)
}

// statusBarrier holds every status call until `n` of them have arrived, then
// releases them together. Two passes that both probed are forced to reach
// MarkInvoicePaid concurrently instead of one of them finishing first, which is
// the only shape in which the zero-row backstop is exercised rather than
// incidentally avoided.
type statusBarrier struct {
	mu      sync.Mutex
	n       int
	arrived int
	ready   chan struct{}
	inner   *statusStubRT
}

func newStatusBarrier(n int, inner *statusStubRT) *statusBarrier {
	return &statusBarrier{n: n, ready: make(chan struct{}), inner: inner}
}

func (b *statusBarrier) RoundTrip(r *http.Request) (*http.Response, error) {
	b.mu.Lock()
	b.arrived++
	if b.arrived >= b.n {
		select {
		case <-b.ready:
		default:
			close(b.ready)
		}
	}
	b.mu.Unlock()
	<-b.ready
	return b.inner.RoundTrip(r)
}

// lockedSeq is a sequence allocator safe for concurrent use, unlike fakeSeq.
type lockedSeq struct {
	mu sync.Mutex
	n  int64
}

func (l *lockedSeq) Sequence(context.Context, string, time.Time, any) (int64, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.n++
	return l.n, nil
}

func derefStr(s *string) string {
	if s == nil {
		return "<nil>"
	}
	return *s
}